style

PeterZh6 · Oct 10, 2024 · ec1df10 · ec1df10
1 parent 00d2757
commit ec1df10
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 10 deletions.
diff --git a/.../src/main/java/org/apache/inlong/sort/pulsar/source/reader/PulsarOrderedSourceReader.java b/.../src/main/java/org/apache/inlong/sort/pulsar/source/reader/PulsarOrderedSourceReader.java
@@ -17,6 +17,9 @@
 
 package org.apache.inlong.sort.pulsar.source.reader;
 
+import org.apache.inlong.sort.base.metric.SourceExactlyMetric;
+import org.apache.inlong.sort.pulsar.table.PulsarTableDeserializationSchema;
+
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.connector.source.ReaderOutput;
@@ -32,8 +35,6 @@
 import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
 import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitState;
 import org.apache.flink.core.io.InputStatus;
-import org.apache.inlong.sort.base.metric.SourceExactlyMetric;
-import org.apache.inlong.sort.pulsar.table.PulsarTableDeserializationSchema;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClient;
@@ -95,8 +96,9 @@ public PulsarOrderedSourceReader(
         this.cursorsOfFinishedSplits = new ConcurrentHashMap<>();
         this.deserializationSchema = deserializationSchema;
         // get SourceExactlyMetric instance from deserializationSchema
-        if(deserializationSchema instanceof PulsarTableDeserializationSchema) {
-            this.sourceExactlyMetric = ((PulsarTableDeserializationSchema)deserializationSchema).getSourceExactlyMetric();
+        if (deserializationSchema instanceof PulsarTableDeserializationSchema) {
+            this.sourceExactlyMetric =
+                    ((PulsarTableDeserializationSchema) deserializationSchema).getSourceExactlyMetric();
         }
     }
 
@@ -167,8 +169,8 @@ public List<PulsarPartitionSplit> snapshotState(long checkpointId) {
             cursors.putAll(cursorsOfFinishedSplits);
 
             return splits;
-        } catch(Exception e) {
-            if(sourceExactlyMetric != null) {
+        } catch (Exception e) {
+            if (sourceExactlyMetric != null) {
                 sourceExactlyMetric.incNumSnapshotError();
                 throw e;
             }

diff --git a/...r/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarTableDeserializationSchema.java b/...r/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarTableDeserializationSchema.java
@@ -17,19 +17,21 @@
 
 package org.apache.inlong.sort.pulsar.table;
 
+import org.apache.inlong.sort.base.metric.MetricOption;
+import org.apache.inlong.sort.base.metric.MetricsCollector;
+import org.apache.inlong.sort.base.metric.SourceExactlyMetric;
+
 import org.apache.flink.api.common.functions.util.ListCollector;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
 import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.util.Collector;
-import org.apache.inlong.sort.base.metric.MetricOption;
-import org.apache.inlong.sort.base.metric.MetricsCollector;
-import org.apache.inlong.sort.base.metric.SourceExactlyMetric;
 import org.apache.pulsar.client.api.Message;
 
 import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -115,7 +117,7 @@ public void deserialize(Message<byte[]> message, Collector<RowData> collector)
 
             rowDataConverter.projectToProducedRowAndCollect(
                     message, keyRowData, valueRowData, metricsCollector);
-            if(sourceExactlyMetric != null) {
+            if (sourceExactlyMetric != null) {
                 sourceExactlyMetric.recordDeserializeDelay(System.currentTimeMillis() - deserializeStartTime);
                 sourceExactlyMetric.incNumDeserializeSuccess();
             }