fixup! [FLINK-37410] Split level Watermark metrics

Author: Efrat19

flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkOutputMultiplexer.java

@@ -17,15 +17,16 @@
  */
 
 package org.apache.flink.api.common.eventtime;
-import static org.apache.flink.util.Preconditions.checkState;
-
-import java.util.HashMap;
-import java.util.Map;
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.eventtime.CombinedWatermarkStatus.PartialWatermark;
 import org.apache.flink.util.Preconditions;
 
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
 /**
  * A {@link WatermarkOutputMultiplexer} combines the watermark (and idleness) updates of multiple
  * partitions/shards/splits into one combined watermark update and forwards it to an underlying

flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/SourceSplitMetricGroup.java

@@ -15,29 +15,28 @@
  * limitations under the License.
  */
 
- package org.apache.flink.metrics.groups;
-
- import org.apache.flink.annotation.PublicEvolving;
- 
- /**
-  * Pre-defined metrics for {@code SplitEnumerator}.
-  *
-  * <p>You should only update the metrics in the main operator thread.
-  */
- @PublicEvolving
- public interface SourceSplitMetricGroup extends OperatorCoordinatorMetricGroup {
-     long getCurrentWatermark();
- 
-     double getActiveTimePerSecond();
- 
-     long getPausedTimePerSecond();
- 
-     long getIdleTimePerSecond();
- 
-     double getAccumulatedActiveTime();
- 
-     long getAccumulatedPausedTime();
- 
-     long getAccumulatedIdleTime();
- }
- 
+package org.apache.flink.metrics.groups;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * Pre-defined metrics for {@code SplitEnumerator}.
+ *
+ * <p>You should only update the metrics in the main operator thread.
+ */
+@PublicEvolving
+public interface SourceSplitMetricGroup extends OperatorCoordinatorMetricGroup {
+    long getCurrentWatermark();
+
+    double getActiveTimePerSecond();
+
+    long getPausedTimePerSecond();
+
+    long getIdleTimePerSecond();
+
+    double getAccumulatedActiveTime();
+
+    long getAccumulatedPausedTime();
+
+    long getAccumulatedIdleTime();
+}

flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java

@@ -178,7 +178,7 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
     private final Map<String, Long> splitCurrentWatermarks = new HashMap<>();
 
     private final Set<String> currentlyPausedSplits = new HashSet<>();
-    
+
     private final Map<String, InternalSourceSplitMetricGroup> splitMetricGroups = new HashMap<>();
 
     private enum OperatingMode {
@@ -775,7 +775,7 @@ private void reportPausedOrResumed(
             getOrCreateSplitMetricGroup(splitId).markPaused();
         }
     }
-    
+
     private void checkWatermarkAlignment() {
         if (operatingMode == OperatingMode.READING) {
             checkState(waitingForAlignmentFuture.isDone());

flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java

@@ -39,13 +39,13 @@ Licensed to the Apache Software Foundation (ASF) under one
 import org.apache.flink.streaming.api.operators.source.CollectingDataOutput;
 import org.apache.flink.streaming.api.operators.source.TestingSourceOperator;
 import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask;
 import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment;
 import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
 import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
 import org.apache.flink.streaming.util.MockOutput;
 import org.apache.flink.streaming.util.MockStreamConfig;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 
 import org.assertj.core.api.Condition;
 import org.junit.jupiter.api.Test;

flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java

@@ -44,7 +44,7 @@ Licensed to the Apache Software Foundation (ASF) under one
 import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
 import org.apache.flink.streaming.util.CollectorOutput;
 import org.apache.flink.util.CollectionUtil;
-import static org.apache.flink.api.common.eventtime.WatermarkMatchers.watermark;
+
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -56,6 +56,7 @@ Licensed to the Apache Software Foundation (ASF) under one
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
+import static org.apache.flink.api.common.eventtime.WatermarkMatchers.watermark;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Unit test for {@link SourceOperator}. */
@@ -270,6 +271,7 @@ public void testMetricGroupTracksSplitWatermark() throws Exception {
                 operator.getSplitMetricGroup(split.splitId()).getCurrentWatermark(),
                 expectedWatermark);
     }
+
     private static class DataOutputToOutput<T> implements PushingAsyncDataInput.DataOutput<T> {
 
         private final Output<StreamRecord<T>> output;