apache · lightzhao · Apr 23, 2024
diff --git a/...ache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReader.java b/...ache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReader.java
@@ -49,6 +49,7 @@ public class AmazonDynamoDBSourceReader
     protected AmazonDynamoDBSourceOptions amazondynamodbSourceOptions;
     protected SeaTunnelRowDeserializer seaTunnelRowDeserializer;
     Queue<AmazonDynamoDBSourceSplit> pendingSplits = new ConcurrentLinkedDeque<>();
+    private static final long SLEEP_TIME_MS = 2000L;
 
     private volatile boolean noMoreSplit;
 
@@ -83,7 +84,6 @@ public void close() {
     }
 
     @Override
-    @SuppressWarnings("magicnumber")
     public void pollNext(Collector<SeaTunnelRow> output) throws InterruptedException {
         synchronized (output.getCheckpointLock()) {
             AmazonDynamoDBSourceSplit split = pendingSplits.poll();
@@ -95,7 +95,7 @@ public void pollNext(Collector<SeaTunnelRow> output) throws InterruptedException
                     // signal to the source that we have reached the end of the data.
                     log.info("Closed the bounded amazonDynamodb source");
                     context.signalNoMoreElement();
-                    Thread.sleep(2000L);
+                    Thread.sleep(SLEEP_TIME_MS);
                 }
             }
             if (Objects.nonNull(split)) {

diff --git a/...org/apache/seatunnel/connectors/cdc/base/relational/connection/JdbcConnectionFactory.java b/...org/apache/seatunnel/connectors/cdc/base/relational/connection/JdbcConnectionFactory.java
@@ -37,6 +37,7 @@ public class JdbcConnectionFactory implements JdbcConnection.ConnectionFactory {
 
     private final JdbcSourceConfig sourceConfig;
     private final JdbcConnectionPoolFactory jdbcConnectionPoolFactory;
+    private static final long SLEEP_TIME_MS = 300L;
 
     public JdbcConnectionFactory(
             JdbcSourceConfig sourceConfig, JdbcConnectionPoolFactory jdbcConnectionPoolFactory) {
@@ -65,7 +66,7 @@ public Connection connect(JdbcConfiguration config) throws SQLException {
             } catch (SQLException e) {
                 if (i < connectRetryTimes - 1) {
                     try {
-                        Thread.sleep(300);
+                        Thread.sleep(SLEEP_TIME_MS);
                     } catch (InterruptedException ie) {
                         throw new SeaTunnelException(
                                 "Failed to get connection, interrupted while doing another attempt",

diff --git a/...unnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java b/...unnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java
@@ -48,6 +48,7 @@ public abstract class AbstractJdbcSourceChunkSplitter implements JdbcSourceChunk
 
     private final JdbcSourceConfig sourceConfig;
     private final JdbcDataSourceDialect dialect;
+    private static final long SLEEP_TIME_MS = 100L;
 
     public AbstractJdbcSourceChunkSplitter(
             JdbcSourceConfig sourceConfig, JdbcDataSourceDialect dialect) {
@@ -435,12 +436,11 @@ protected int ObjectCompare(Object obj1, Object obj2) {
         return ObjectUtils.compare(obj1, obj2);
     }
 
-    @SuppressWarnings("MagicNumber")
     private static void maySleep(int count, TableId tableId) {
         // every 100 queries to sleep 1s
         if (count % 10 == 0) {
             try {
-                Thread.sleep(100);
+                Thread.sleep(SLEEP_TIME_MS);
             } catch (InterruptedException e) {
                 // nothing to do
             }

diff --git a/.../main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java b/.../main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java
@@ -78,6 +78,7 @@ public class PostgresReplicationConnection extends JdbcConnection implements Rep
     private Lsn defaultStartingPos;
     private SlotCreationResult slotCreationInfo;
     private boolean hasInitedSlot;
+    private static final long SLEEP_TIME_MS = 10L;
 
     /**
      * Creates a new replication connection with the given params.
@@ -721,7 +722,7 @@ private PGReplicationStream startPgReplicationStream(
         // TODO DBZ-508 get rid of this
         // Needed by tests when connections are opened and closed in a fast sequence
         try {
-            Thread.sleep(10);
+            Thread.sleep(SLEEP_TIME_MS);
         } catch (Exception e) {
         }
         stream.forceUpdateStatus();

diff --git a/...org/apache/seatunnel/connectors/seatunnel/common/multitablesink/MultiTableSinkWriter.java b/...org/apache/seatunnel/connectors/seatunnel/common/multitablesink/MultiTableSinkWriter.java
@@ -53,6 +53,7 @@ public class MultiTableSinkWriter
     private final ExecutorService executorService;
     private MultiTableResourceManager resourceManager;
     private volatile boolean submitted = false;
+    private static final long SLEEP_TIME_MS = 100L;
 
     public MultiTableSinkWriter(
             Map<SinkIdentifier, SinkWriter<SeaTunnelRow, ?, ?>> sinkWriters, int queueSize) {
@@ -305,7 +306,7 @@ private void checkQueueRemain() {
         try {
             for (BlockingQueue<SeaTunnelRow> blockingQueue : blockingQueues) {
                 while (!blockingQueue.isEmpty()) {
-                    Thread.sleep(100);
+                    Thread.sleep(SLEEP_TIME_MS);
                     subSinkErrorCheck();
                 }
             }

diff --git a/...java/org/apache/seatunnel/connectors/seatunnel/common/source/reader/SourceReaderBase.java b/...java/org/apache/seatunnel/connectors/seatunnel/common/source/reader/SourceReaderBase.java
@@ -66,6 +66,7 @@ public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitSt
     protected SplitContext<T, SplitStateT> currentSplitContext;
     private Collector<T> currentSplitOutput;
     private boolean noMoreSplitsAssignment;
+    private static final long SLEEP_TIME_MS = 100L;
 
     public SourceReaderBase(
             BlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
@@ -162,7 +163,7 @@ private RecordsWithSplitIds<E> getNextFetch(Collector<T> output) {
         if (recordsWithSplitId == null || !moveToNextSplit(recordsWithSplitId, output)) {
             try {
                 log.trace("Current fetch is finished.");
-                Thread.sleep(100);
+                Thread.sleep(SLEEP_TIME_MS);
             } catch (InterruptedException e) {
                 throw new SeaTunnelException(e);
             }

diff --git a/...s/src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReader.java b/...s/src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReader.java
@@ -70,6 +70,7 @@ public class DorisValueReader {
 
     protected SeaTunnelRowType seaTunnelRowType;
     protected boolean asyncThreadStarted;
+    private static final long SLEEP_TIME_MS = 5L;
 
     public DorisValueReader(
             PartitionDefinition partition, DorisConfig config, SeaTunnelRowType seaTunnelRowType) {
@@ -205,7 +206,7 @@ public boolean hasNext() {
                     } else {
                         // wait for rowBatch put in queue or eos change
                         try {
-                            Thread.sleep(5);
+                            Thread.sleep(SLEEP_TIME_MS);
                         } catch (InterruptedException e) {
                         }
                     }

diff --git a/...src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java b/...src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java
@@ -48,6 +48,7 @@ public class FakeSourceReader implements SourceReader<SeaTunnelRow, FakeSourceSp
     private volatile boolean noMoreSplit;
     private final long minSplitReadInterval;
     private volatile long latestTimestamp = 0;
+    private static final long SLEEP_TIME_MS = 1000L;
 
     public FakeSourceReader(
             SourceReader.Context context,
@@ -83,7 +84,6 @@ public void close() {
     }
 
     @Override
-    @SuppressWarnings("MagicNumber")
     public void pollNext(Collector<SeaTunnelRow> output) throws InterruptedException {
         long currentTimestamp = Instant.now().toEpochMilli();
         if (currentTimestamp <= latestTimestamp + minSplitReadInterval) {
@@ -117,7 +117,7 @@ public void pollNext(Collector<SeaTunnelRow> output) throws InterruptedException
             log.info("Closed the bounded fake source");
             context.signalNoMoreElement();
         }
-        Thread.sleep(1000L);
+        Thread.sleep(SLEEP_TIME_MS);
     }
 
     @Override

diff --git a/...main/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSourceReader.java b/...main/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSourceReader.java
@@ -38,6 +38,7 @@ public class BaseFileSourceReader implements SourceReader<SeaTunnelRow, FileSour
     private final SourceReader.Context context;
     private final Deque<FileSourceSplit> sourceSplits = new ConcurrentLinkedDeque<>();
     private volatile boolean noMoreSplit;
+    private static final long SLEEP_TIME_MS = 1000L;
 
     public BaseFileSourceReader(ReadStrategy readStrategy, SourceReader.Context context) {
         this.readStrategy = readStrategy;
@@ -69,7 +70,7 @@ public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
                 log.info("Closed the bounded File source");
                 context.signalNoMoreElement();
             } else {
-                Thread.sleep(1000L);
+                Thread.sleep(SLEEP_TIME_MS);
             }
         }
     }

diff --git a/...src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java b/...src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java
@@ -67,6 +67,7 @@ public class HttpSourceReader extends AbstractSingleSplitReader<SeaTunnelRow> {
             Configuration.defaultConfiguration().addOptions(DEFAULT_OPTIONS);
     private boolean noMoreElementFlag = true;
     private Optional<PageInfo> pageInfoOptional = Optional.empty();
+    private static final long SLEEP_TIME_MS = 10L;
 
     public HttpSourceReader(
             HttpParameter httpParameter,
@@ -167,7 +168,7 @@ public void internalPollNext(Collector<SeaTunnelRow> output) throws Exception {
                     updateRequestParam(info);
                     pollAndCollectData(output);
                     pageIndex += 1;
-                    Thread.sleep(10);
+                    Thread.sleep(SLEEP_TIME_MS);
                 }
             } else {
                 pollAndCollectData(output);

diff --git a/...main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/DynamicChunkSplitter.java b/...main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/DynamicChunkSplitter.java
@@ -54,6 +54,8 @@
 @Slf4j
 public class DynamicChunkSplitter extends ChunkSplitter {
 
+    private static final long SLEEP_TIME_MS = 100L;
+
     public DynamicChunkSplitter(JdbcSourceConfig config) {
         super(config);
     }
@@ -439,7 +441,7 @@ private static void maySleep(int count, TablePath tablePath) {
         // every 100 queries to sleep 1s
         if (count % 10 == 0) {
             try {
-                Thread.sleep(100);
+                Thread.sleep(SLEEP_TIME_MS);
             } catch (InterruptedException e) {
                 // nothing to do
             }

diff --git a/...src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceReader.java b/...src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceReader.java
@@ -40,6 +40,7 @@ public class JdbcSourceReader implements SourceReader<SeaTunnelRow, JdbcSourceSp
     private final JdbcInputFormat inputFormat;
     private final Deque<JdbcSourceSplit> splits = new ConcurrentLinkedDeque<>();
     private volatile boolean noMoreSplit;
+    private static final long SLEEP_TIME_MS = 1000L;
 
     public JdbcSourceReader(
             Context context, JdbcSourceConfig config, Map<TablePath, CatalogTable> tables) {
@@ -58,7 +59,6 @@ public void close() throws IOException {
     }
 
     @Override
-    @SuppressWarnings("magicnumber")
     public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
         synchronized (output.getCheckpointLock()) {
             JdbcSourceSplit split = splits.poll();
@@ -77,7 +77,7 @@ public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
                 log.info("Closed the bounded jdbc source");
                 context.signalNoMoreElement();
             } else {
-                Thread.sleep(1000L);
+                Thread.sleep(SLEEP_TIME_MS);
             }
         }
     }

diff --git a/...src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceReader.java b/...src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceReader.java
@@ -49,6 +49,7 @@ public class KuduSourceReader implements SourceReader<SeaTunnelRow, KuduSourceSp
     boolean noMoreSplit;
 
     private final Map<TablePath, SeaTunnelRowType> tables;
+    private static final long SLEEP_TIME_MS = 1000L;
 
     public KuduSourceReader(Context context, KuduSourceConfig kuduSourceConfig) {
         this.context = context;
@@ -99,7 +100,7 @@ public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
                 log.info("Closed the bounded kudu source");
                 context.signalNoMoreElement();
             } else {
-                Thread.sleep(1000L);
+                Thread.sleep(SLEEP_TIME_MS);
             }
         }
     }

diff --git a/...main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceReader.java b/...main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceReader.java
@@ -46,6 +46,7 @@ public class PaimonSourceReader implements SourceReader<SeaTunnelRow, PaimonSour
     private final Table table;
     private final SeaTunnelRowType seaTunnelRowType;
     private volatile boolean noMoreSplit;
+    private static final long SLEEP_TIME_MS = 1000L;
 
     public PaimonSourceReader(Context context, Table table, SeaTunnelRowType seaTunnelRowType) {
         this.context = context;
@@ -86,7 +87,7 @@ public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
                 context.signalNoMoreElement();
             } else {
                 log.warn("Waiting for flink table source split, sleeping 1s");
-                Thread.sleep(1000L);
+                Thread.sleep(SLEEP_TIME_MS);
             }
         }
     }

diff --git a/...l-e2e-common/src/test/java/org/apache/seatunnel/e2e/sink/inmemory/InMemorySinkWriter.java b/...l-e2e-common/src/test/java/org/apache/seatunnel/e2e/sink/inmemory/InMemorySinkWriter.java
@@ -38,6 +38,7 @@ public class InMemorySinkWriter
 
     // use a daemon thread to test classloader leak
     private static final Thread THREAD;
+    private static final long SLEEP_TIME_MS = 1000L;
 
     static {
         // use the daemon thread to always hold the classloader
@@ -47,7 +48,7 @@ public class InMemorySinkWriter
                         () -> {
                             while (true) {
                                 try {
-                                    Thread.sleep(1000);
+                                    Thread.sleep(SLEEP_TIME_MS);
                                     System.out.println(classLoader);
                                 } catch (InterruptedException e) {
                                     e.printStackTrace();

diff --git a/...er/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java b/...er/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
@@ -147,6 +147,7 @@ public class CheckpointCoordinator {
     private PendingCheckpoint savepointPendingCheckpoint;
 
     private final String checkpointStateImapKey;
+    private static final long SLEEP_TIME_MS = 500L;
 
     @SneakyThrows
     public CheckpointCoordinator(
@@ -490,7 +491,7 @@ public PassiveCompletableFuture<CompletedCheckpoint> startSavepoint() {
         CompletableFuture<PendingCheckpoint> savepoint;
         synchronized (lock) {
             while (pendingCounter.get() > 0 && !shutdown) {
-                Thread.sleep(500);
+                Thread.sleep(SLEEP_TIME_MS);
             }
             if (shutdown || isCompleted()) {
                 return completableFutureWithError(

diff --git a/...-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java b/...-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java
@@ -118,6 +118,8 @@ public class PhysicalVertex {
 
     public volatile boolean isRunning = false;
 
+    private static final long SLEEP_TIME_MS = 2000L;
+
     /** The error throw by physicalVertex, should be set when physicalVertex throw error. */
     private AtomicReference<String> errorByPhysicalVertex = new AtomicReference<>();
 
@@ -441,7 +443,7 @@ private void noticeTaskExecutionServiceCancel() {
                                 "%s cancel failed with Exception: %s, retry %s",
                                 this.getTaskFullName(), ExceptionUtils.getMessage(e), i));
                 try {
-                    Thread.sleep(2000);
+                    Thread.sleep(SLEEP_TIME_MS);
                 } catch (InterruptedException ex) {
                     throw new RuntimeException(ex);
                 }

diff --git a/...el-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelTask.java b/...el-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelTask.java
@@ -115,6 +115,8 @@ public abstract class SeaTunnelTask extends AbstractTask {
 
     private SeaTunnelMetricsContext metricsContext;
 
+    private static final long SLEEP_TIME_MS = 100L;
+
     public SeaTunnelTask(long jobID, TaskLocation taskID, int indexID, Flow executionFlow) {
         super(jobID, taskID);
         this.indexID = indexID;
@@ -151,14 +153,14 @@ protected void stateProcess() throws Exception {
                     currState = READY_START;
                     reportTaskStatus(READY_START);
                 } else {
-                    Thread.sleep(100);
+                    Thread.sleep(SLEEP_TIME_MS);
                 }
                 break;
             case READY_START:
                 if (startCalled) {
                     currState = STARTING;
                 } else {
-                    Thread.sleep(100);
+                    Thread.sleep(SLEEP_TIME_MS);
                 }
                 break;
             case STARTING:
@@ -174,7 +176,7 @@ protected void stateProcess() throws Exception {
                 if (closeCalled) {
                     currState = CLOSED;
                 } else {
-                    Thread.sleep(100);
+                    Thread.sleep(SLEEP_TIME_MS);
                 }
                 break;
             case CLOSED: