[FLINK-33128] Add converter.open() method on TestValuesRuntimeFunctions

This closes #23453.
apache · Oct 26, 2023 · f31770f · f31770f
1 parent 0388b76
commit f31770f
Showing 1 changed file with 7 additions and 6 deletions.
diff --git a/...er/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java b/...er/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java
@@ -36,6 +36,7 @@
 import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.connector.RuntimeConverter;
 import org.apache.flink.table.connector.sink.DynamicTableSink.DataStructureConverter;
 import org.apache.flink.table.connector.source.LookupTableSource;
 import org.apache.flink.table.data.GenericRowData;
@@ -631,11 +632,11 @@ protected TestValuesLookupFunction(
         public void open(FunctionContext context) throws Exception {
             RESOURCE_COUNTER.incrementAndGet();
             isOpenCalled = true;
+            ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
             if (projectable) {
-                projection =
-                        generatedProjection.newInstance(
-                                Thread.currentThread().getContextClassLoader());
+                projection = generatedProjection.newInstance(classLoader);
             }
+            converter.open(RuntimeConverter.Context.create(classLoader));
             rowSerializer = InternalSerializers.create(producedRowType);
             indexDataByKey();
         }
@@ -725,11 +726,11 @@ protected AsyncTestValueLookupFunction(
         @Override
         public void open(FunctionContext context) throws Exception {
             RESOURCE_COUNTER.incrementAndGet();
+            ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
             if (projectable) {
-                projection =
-                        generatedProjection.newInstance(
-                                Thread.currentThread().getContextClassLoader());
+                projection = generatedProjection.newInstance(classLoader);
             }
+            converter.open(RuntimeConverter.Context.create(classLoader));
             rowSerializer = InternalSerializers.create(producedRowType);
             isOpenCalled = true;
             // generate unordered result for async lookup