[FLINK-21549][table-planner-blink] Remove DynamicTableSinkSpecJsonDes…

…erializer and set ClassLoader and Configuration to DynamicTableSinkSpec when creating ExecNodeGraph This closes apache#15062
derekm · Mar 7, 2021 · b696144 · b696144
1 parent df72f25
commit b696144
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 60 deletions.
diff --git a/...k/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java b/...k/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
@@ -48,7 +48,6 @@
 import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
 import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
 import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
-import org.apache.flink.table.planner.plan.nodes.exec.serde.DynamicTableSinkSpecJsonDeserializer;
 import org.apache.flink.table.planner.plan.nodes.exec.spec.DynamicTableSinkSpec;
 import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
 import org.apache.flink.table.planner.sinks.TableSinkUtils;
@@ -64,7 +63,6 @@
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 
 import java.util.Arrays;
 import java.util.List;
@@ -81,7 +79,6 @@ public abstract class CommonExecSink extends ExecNodeBase<Object> {
  public static final String FIELD_NAME_DYNAMIC_TABLE_SINK = "dynamicTableSink";
 
  @JsonProperty(FIELD_NAME_DYNAMIC_TABLE_SINK)
- @JsonDeserialize(using = DynamicTableSinkSpecJsonDeserializer.class)
  protected final DynamicTableSinkSpec tableSinkSpec;
 
  @JsonIgnore private final ChangelogMode changelogMode;
@@ -101,6 +98,10 @@ protected CommonExecSink(
  this.isBounded = isBounded;
  }
 
+ public DynamicTableSinkSpec getTableSinkSpec() {
+ return tableSinkSpec;
+ }
+
  protected Transformation<Object> createSinkTransformation(
  StreamExecutionEnvironment env,
  TableConfig tableConfig,

diff --git a/...pache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSinkSpecJsonDeserializer.java b/...pache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSinkSpecJsonDeserializer.java
diff --git a/.../org/apache/flink/table/planner/plan/nodes/exec/serde/ExecNodeGraphJsonPlanGenerator.java b/.../org/apache/flink/table/planner/plan/nodes/exec/serde/ExecNodeGraphJsonPlanGenerator.java
@@ -26,6 +26,8 @@
 import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
 import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
 import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeGraph;
+import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecSink;
+import org.apache.flink.table.planner.plan.nodes.exec.spec.DynamicTableSinkSpec;
 import org.apache.flink.table.planner.plan.nodes.exec.spec.DynamicTableSourceSpec;
 import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecTableSourceScan;
 import org.apache.flink.table.planner.plan.nodes.exec.visitor.AbstractExecNodeExactlyOnceVisitor;
@@ -253,6 +255,11 @@ public ExecNodeGraph convertToExecNodeGraph(SerdeContext serdeCtx) {
  tableSourceSpec.setReadableConfig(serdeCtx.getConfiguration());
  tableSourceSpec.setClassLoader(serdeCtx.getClassLoader());
  applyProjectionPushDown((StreamExecTableSourceScan) execNode);
+ } else if (execNode instanceof CommonExecSink) {
+ DynamicTableSinkSpec tableSinkSpec =
+ ((CommonExecSink) execNode).getTableSinkSpec();
+ tableSinkSpec.setReadableConfig(serdeCtx.getConfiguration());
+ tableSinkSpec.setClassLoader(serdeCtx.getClassLoader());
  }
  idToExecNodes.put(id, execNode);
  }

diff --git a/...a/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSinkSpecSerdeTest.java b/...a/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSinkSpecSerdeTest.java
@@ -44,7 +44,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertNull;
 
 /** Tests for {@link DynamicTableSinkSpec} serialization and deserialization. */
 @RunWith(Parameterized.class)
@@ -63,8 +63,6 @@ public void testDynamicTableSinkSpecSerde() throws IOException {
  FlinkSqlOperatorTable.instance());
  ObjectMapper mapper = JsonSerdeUtil.createObjectMapper(serdeCtx);
  SimpleModule module = new SimpleModule();
- module.addDeserializer(
- DynamicTableSinkSpec.class, new DynamicTableSinkSpecJsonDeserializer());
  mapper.registerModule(module);
  StringWriter writer = new StringWriter(100);
  try (JsonGenerator gen = mapper.getFactory().createGenerator(writer)) {
@@ -73,7 +71,10 @@ public void testDynamicTableSinkSpecSerde() throws IOException {
  String json = writer.toString();
  DynamicTableSinkSpec actual = mapper.readValue(json, DynamicTableSinkSpec.class);
  assertEquals(spec, actual);
- assertSame(classLoader, actual.getClassLoader());
+ assertNull(actual.getReadableConfig());
+ actual.setReadableConfig(serdeCtx.getConfiguration());
+ assertNull(actual.getClassLoader());
+ actual.setClassLoader(serdeCtx.getClassLoader());
  assertNotNull(actual.getTableSink());
  }