[FLINK-21442][connectors/jdbc] Fix XaSinkStateSerializer serialization

deepfield · Feb 25, 2021 · eb0c19d · eb0c19d
1 parent 7fd4767
commit eb0c19d
Showing 1 changed file with 26 additions and 19 deletions.
diff --git a/...onnector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateSerializer.java b/...onnector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateSerializer.java
@@ -32,30 +32,14 @@
 
 /** XaSinkStateSerializer. */
 @Internal
-final class XaSinkStateSerializer extends TypeSerializer<JdbcXaSinkFunctionState> {
+public final class XaSinkStateSerializer extends TypeSerializer<JdbcXaSinkFunctionState> {
 
  private static final TypeSerializerSnapshot<JdbcXaSinkFunctionState> SNAPSHOT =
- new SimpleTypeSerializerSnapshot<JdbcXaSinkFunctionState>(XaSinkStateSerializer::new) {
- private static final int VERSION = 1;
-
- @Override
- public void writeSnapshot(DataOutputView out) throws IOException {
- super.writeSnapshot(out);
- out.writeInt(VERSION);
- }
-
- @Override
- public void readSnapshot(int readVersion, DataInputView in, ClassLoader classLoader)
- throws IOException {
- super.readSnapshot(readVersion, in, classLoader);
- in.readInt();
- }
- };
-
+ new XaSinkStateSimpleXaTypeSerializerSnapshot();
  private final TypeSerializer<Xid> xidSerializer;
  private final TypeSerializer<CheckpointAndXid> checkpointAndXidSerializer;
 
- XaSinkStateSerializer() {
+ public XaSinkStateSerializer() {
  this(new XidSerializer(), new CheckpointAndXidSerializer());
  }
 
@@ -149,4 +133,27 @@ public int hashCode() {
  public TypeSerializerSnapshot<JdbcXaSinkFunctionState> snapshotConfiguration() {
  return SNAPSHOT;
  }
+
+ /** Simple {@link TypeSerializerSnapshot} for {@link XaSinkStateSerializer}. */
+ public static class XaSinkStateSimpleXaTypeSerializerSnapshot
+ extends SimpleTypeSerializerSnapshot<JdbcXaSinkFunctionState> {
+ private static final int VERSION = 1;
+
+ public XaSinkStateSimpleXaTypeSerializerSnapshot() {
+ super(XaSinkStateSerializer::new);
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {
+ super.writeSnapshot(out);
+ out.writeInt(VERSION);
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader classLoader)
+ throws IOException {
+ super.readSnapshot(readVersion, in, classLoader);
+ in.readInt();
+ }
+ }
 }