[FLINK-34316] Reduce instantiation of ScanRuntimeProvider in streamin…

…g mode
apache · Jan 31, 2024 · 5035808 · 5035808
1 parent ff9a3c8
commit 5035808
Showing 1 changed file with 4 additions and 4 deletions.
diff --git a/...e-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSourceUtils.java b/...e-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSourceUtils.java
@@ -498,14 +498,12 @@ private static void validateScanSource(
             ScanTableSource scanSource,
             boolean isBatchMode,
             ReadableConfig config) {
-        final ScanRuntimeProvider provider =
-                scanSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE);
         final ChangelogMode changelogMode = scanSource.getChangelogMode();
 
         validateWatermarks(tableDebugName, schema);
 
         if (isBatchMode) {
-            validateScanSourceForBatch(tableDebugName, changelogMode, provider);
+            validateScanSourceForBatch(tableDebugName, scanSource, changelogMode);
         } else {
             validateScanSourceForStreaming(
                     tableDebugName, schema, scanSource, changelogMode, config);
@@ -558,7 +556,9 @@ private static void validateScanSourceForStreaming(
     }
 
     private static void validateScanSourceForBatch(
-            String tableDebugName, ChangelogMode changelogMode, ScanRuntimeProvider provider) {
+            String tableDebugName, ScanTableSource scanSource, ChangelogMode changelogMode) {
+        final ScanRuntimeProvider provider =
+                scanSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE);
         // batch only supports bounded source
         if (!provider.isBounded()) {
             throw new ValidationException(