fix style

apache · Aug 28, 2023 · 7bdb2b0 · 7bdb2b0
1 parent 6107ac3
commit 7bdb2b0
Show file tree

Hide file tree

Showing 6 changed files with 19 additions and 15 deletions.
diff --git a/...v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatistics.java b/...v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatistics.java
@@ -24,9 +24,9 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
 /**
- * AggregatedStatistics is used by {@link DataStatisticsCoordinator} to collect {@link DataStatistics}
- * from {@link DataStatisticsOperator} subtasks for specific checkpoint. It stores the merged {@link
- * DataStatistics} result from all reported subtasks.
+ * AggregatedStatistics is used by {@link DataStatisticsCoordinator} to collect {@link
+ * DataStatistics} from {@link DataStatisticsOperator} subtasks for specific checkpoint. It stores
+ * the merged {@link DataStatistics} result from all reported subtasks.
  */
 class AggregatedStatistics<D extends DataStatistics<D, S>, S> implements Serializable {
 

diff --git a/.../flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java b/.../flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java
@@ -46,10 +46,9 @@
 /**
  * DataStatisticsCoordinator receives {@link DataStatisticsEvent} from {@link
  * DataStatisticsOperator} every subtask and then merge them together. Once aggregation for all
- * subtasks data statistics completes, DataStatisticsCoordinator will send the aggregated
- * data statistics back to {@link DataStatisticsOperator}. In the end a custom
- * partitioner will distribute traffic based on the aggregated data statistics to improve data
- * clustering.
+ * subtasks data statistics completes, DataStatisticsCoordinator will send the aggregated data
+ * statistics back to {@link DataStatisticsOperator}. In the end a custom partitioner will
+ * distribute traffic based on the aggregated data statistics to improve data clustering.
  */
 @Internal
 class DataStatisticsCoordinator<D extends DataStatistics<D, S>, S> implements OperatorCoordinator {

diff --git a/....17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java b/....17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
@@ -115,7 +115,10 @@ public void handleOperatorEvent(OperatorEvent event) {
             "Operator %s subtask %s received unexpected operator event %s",
             operatorName, subtaskIndex, event.getClass()));
     DataStatisticsEvent<D, S> statisticsEvent = (DataStatisticsEvent<D, S>) event;
-    LOG.info("Operator {} received global data event from coordinator checkpoint {}", operatorName, statisticsEvent.checkpointId());
+    LOG.info(
+        "Operator {} received global data event from coordinator checkpoint {}",
+        operatorName,
+        statisticsEvent.checkpointId());
     globalStatistics =
         DataStatisticsUtil.deserializeDataStatistics(
             statisticsEvent.statisticsBytes(), statisticsSerializer);

diff --git a/....17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOrRecord.java b/....17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOrRecord.java
@@ -43,8 +43,7 @@ class DataStatisticsOrRecord<D extends DataStatistics<D, S>, S> implements Seria
 
   private DataStatisticsOrRecord(DataStatistics<D, S> statistics, RowData record) {
     Preconditions.checkArgument(
-        record != null ^ statistics != null,
-        "DataStatistics or record, not neither or both");
+        record != null ^ statistics != null, "DataStatistics or record, not neither or both");
     this.statistics = statistics;
     this.record = record;
   }

diff --git a/...k/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsUtil.java b/...k/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsUtil.java
@@ -77,9 +77,10 @@ static <D extends DataStatistics<D, S>, S> byte[] serializeAggregatedStatistics(
   }
 
   @SuppressWarnings("unchecked")
-  static <D extends DataStatistics<D, S>, S> AggregatedStatistics<D, S> deserializeAggregatedStatistics(
-      byte[] bytes, TypeSerializer<DataStatistics<D, S>> statisticsSerializer)
-      throws IOException {
+  static <D extends DataStatistics<D, S>, S>
+      AggregatedStatistics<D, S> deserializeAggregatedStatistics(
+          byte[] bytes, TypeSerializer<DataStatistics<D, S>> statisticsSerializer)
+          throws IOException {
     ByteArrayInputStream bytesIn = new ByteArrayInputStream(bytes);
     ObjectInputStream in = new ObjectInputStream(bytesIn);
 

diff --git a/.../src/test/java/org/apache/iceberg/flink/sink/shuffle/TestAggregatedStatisticsTracker.java b/.../src/test/java/org/apache/iceberg/flink/sink/shuffle/TestAggregatedStatisticsTracker.java
@@ -43,7 +43,8 @@ public class TestAggregatedStatisticsTracker {
   private final TypeSerializer<RowData> rowSerializer = new RowDataSerializer(rowType);
   private final TypeSerializer<DataStatistics<MapDataStatistics, Map<RowData, Long>>>
       statisticsSerializer = MapDataStatisticsSerializer.fromKeySerializer(rowSerializer);
-  private AggregatedStatisticsTracker<MapDataStatistics, Map<RowData, Long>> aggregatedStatisticsTracker;
+  private AggregatedStatisticsTracker<MapDataStatistics, Map<RowData, Long>>
+      aggregatedStatisticsTracker;
 
   @Before
   public void before() throws Exception {
@@ -98,7 +99,8 @@ public void receiveOlderDataStatisticEventTest() {
     Assert.assertNull(
         aggregatedStatisticsTracker.receiveDataStatisticEventAndCheckCompletion(
             1, checkpoint1Subtask1DataStatisticEvent));
-    // Receive event from old checkpoint, aggregatedStatisticsAggregatorTracker lastCompletedAggregator
+    // Receive event from old checkpoint, aggregatedStatisticsAggregatorTracker
+    // lastCompletedAggregator
     // and inProgressAggregator won't be updated
     Assert.assertEquals(2, aggregatedStatisticsTracker.inProgressStatistics().checkpointId());
   }