[FLINK-32222] Automatically flush outputStreams in serializers

apache · Jun 8, 2023 · 4f1cdfa · 4f1cdfa
1 parent 7919a75
commit 4f1cdfa
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 11 deletions.
diff --git a/...pache/flink/connector/cassandra/source/enumerator/CassandraEnumeratorStateSerializer.java b/...pache/flink/connector/cassandra/source/enumerator/CassandraEnumeratorStateSerializer.java
@@ -49,9 +49,9 @@ public int getVersion() {
 
     @Override
     public byte[] serialize(CassandraEnumeratorState cassandraEnumeratorState) throws IOException {
-        try (final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
-                final ObjectOutputStream objectOutputStream =
-                        new ObjectOutputStream(byteArrayOutputStream)) {
+        final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+        try (final ObjectOutputStream objectOutputStream =
+                new ObjectOutputStream(byteArrayOutputStream)) {
             final Queue<CassandraSplit> splitsToReassign =
                     cassandraEnumeratorState.getSplitsToReassign();
             objectOutputStream.writeInt(splitsToReassign.size());
@@ -69,10 +69,8 @@ public byte[] serialize(CassandraEnumeratorState cassandraEnumeratorState) throw
                     cassandraEnumeratorState.getStartToken(), objectOutputStream);
             BigIntegerSerializationUtils.write(
                     cassandraEnumeratorState.getMaxToken(), objectOutputStream);
-
-            objectOutputStream.flush();
-            return byteArrayOutputStream.toByteArray();
         }
+        return byteArrayOutputStream.toByteArray();
     }
 
     @Override

diff --git a/...main/java/org/apache/flink/connector/cassandra/source/split/CassandraSplitSerializer.java b/...main/java/org/apache/flink/connector/cassandra/source/split/CassandraSplitSerializer.java
@@ -44,16 +44,15 @@ public int getVersion() {
 
     @Override
     public byte[] serialize(CassandraSplit cassandraSplit) throws IOException {
-        try (final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
-                final ObjectOutputStream objectOutputStream =
-                        new ObjectOutputStream(byteArrayOutputStream)) {
+        final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+        try (final ObjectOutputStream objectOutputStream =
+                new ObjectOutputStream(byteArrayOutputStream)) {
             BigIntegerSerializationUtils.write(
                     cassandraSplit.getRingRangeStart(), objectOutputStream);
             BigIntegerSerializationUtils.write(
                     cassandraSplit.getRingRangeEnd(), objectOutputStream);
-            objectOutputStream.flush();
-            return byteArrayOutputStream.toByteArray();
         }
+        return byteArrayOutputStream.toByteArray();
     }
 
     @Override