[Hotfix][Connector/JDBC] Remove the unused 'offset' attribute of JdbcSourceSplit

Author: RocMarshal

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/enumerator/SqlTemplateSplitEnumerator.java

@@ -54,8 +54,7 @@ public List<JdbcSourceSplit> enumerateSplits() throws IOException {
 
         if (parameterValuesProvider == null) {
             return Collections.singletonList(
-                    new JdbcSourceSplit(
-                            getNextId(), sqlTemplate, null, 0, new CheckpointedOffset()));
+                    new JdbcSourceSplit(getNextId(), sqlTemplate, null, new CheckpointedOffset()));
         }
 
         if (optionalSqlSplitEnumeratorState != null) {
@@ -68,15 +67,14 @@ public List<JdbcSourceSplit> enumerateSplits() throws IOException {
 
         if (parameters == null) {
             return Collections.singletonList(
-                    new JdbcSourceSplit(
-                            getNextId(), sqlTemplate, null, 0, new CheckpointedOffset()));
+                    new JdbcSourceSplit(getNextId(), sqlTemplate, null, new CheckpointedOffset()));
         }
 
         List<JdbcSourceSplit> jdbcSourceSplitList = new ArrayList<>(parameters.length);
         for (Serializable[] paramArr : parameters) {
             jdbcSourceSplitList.add(
                     new JdbcSourceSplit(
-                            getNextId(), sqlTemplate, paramArr, 0, new CheckpointedOffset()));
+                            getNextId(), sqlTemplate, paramArr, new CheckpointedOffset()));
         }
         return jdbcSourceSplitList;
     }

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/split/JdbcSourceSplit.java

@@ -39,35 +39,27 @@ public class JdbcSourceSplit implements SourceSplit, Serializable {
 
     private final @Nullable Serializable[] parameters;
 
-    private final int offset;
-
     private final @Nullable CheckpointedOffset checkpointedOffset;
 
     public JdbcSourceSplit(
             String id,
             String sqlTemplate,
             @Nullable Serializable[] parameters,
-            int offset,
             @Nullable CheckpointedOffset checkpointedOffset) {
         this.id = id;
         this.sqlTemplate = sqlTemplate;
         this.parameters = parameters;
-        this.offset = offset;
         this.checkpointedOffset = checkpointedOffset;
     }
 
-    public int getOffset() {
-        return offset;
-    }
-
     @Nullable
     public CheckpointedOffset getCheckpointedOffset() {
         return checkpointedOffset;
     }
 
     public JdbcSourceSplit updateWithCheckpointedPosition(
             @Nullable CheckpointedOffset checkpointedOffset) {
-        return new JdbcSourceSplit(id, sqlTemplate, parameters, offset, checkpointedOffset);
+        return new JdbcSourceSplit(id, sqlTemplate, parameters, checkpointedOffset);
     }
 
     public Optional<CheckpointedOffset> getReaderPositionOptional() {
@@ -105,16 +97,15 @@ public boolean equals(Object o) {
             return false;
         }
         JdbcSourceSplit that = (JdbcSourceSplit) o;
-        return offset == that.offset
-                && Objects.equals(id, that.id)
+        return Objects.equals(id, that.id)
                 && Objects.equals(sqlTemplate, that.sqlTemplate)
                 && Arrays.equals(parameters, that.parameters)
                 && Objects.equals(checkpointedOffset, that.checkpointedOffset);
     }
 
     @Override
     public int hashCode() {
-        int result = Objects.hash(id, sqlTemplate, offset, checkpointedOffset);
+        int result = Objects.hash(id, sqlTemplate, checkpointedOffset);
         result = 31 * result + Arrays.hashCode(parameters);
         return result;
     }
@@ -130,8 +121,6 @@ public String toString() {
                 + '\''
                 + ", parameters="
                 + Arrays.toString(parameters)
-                + ", offset="
-                + offset
                 + ", checkpointedOffset="
                 + checkpointedOffset
                 + '}';

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/split/JdbcSourceSplitSerializer.java

@@ -81,8 +81,6 @@ public void serializeJdbcSourceSplit(DataOutputStream out, JdbcSourceSplit sourc
         out.writeInt(paramsBytes.length);
         out.write(paramsBytes);
 
-        out.writeInt(sourceSplit.getOffset());
-
         CheckpointedOffset checkpointedOffset = sourceSplit.getCheckpointedOffset();
         byte[] chkOffset = InstantiationUtil.serializeObject(checkpointedOffset);
         out.writeInt(chkOffset.length);
@@ -100,14 +98,12 @@ public JdbcSourceSplit deserializeJdbcSourceSplit(DataInputStream in)
                 InstantiationUtil.deserializeObject(
                         parametersBytes, in.getClass().getClassLoader());
 
-        int offset = in.readInt();
-
         int chkOffsetBytesLen = in.readInt();
         byte[] chkOffsetBytes = new byte[chkOffsetBytesLen];
         in.read(chkOffsetBytes);
         CheckpointedOffset chkOffset =
                 InstantiationUtil.deserializeObject(chkOffsetBytes, in.getClass().getClassLoader());
 
-        return new JdbcSourceSplit(id, sqlTemplate, params, offset, chkOffset);
+        return new JdbcSourceSplit(id, sqlTemplate, params, chkOffset);
     }
 }

flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/source/enumerator/JdbcSourceEnumStateSerializerTest.java

@@ -37,15 +37,14 @@ class JdbcSourceEnumStateSerializerTest {
 
     private final JdbcSourceEnumeratorState state =
             new JdbcSourceEnumeratorState(
-                    Arrays.asList(new JdbcSourceSplit("1", "select 1", null, 0, null)),
+                    Arrays.asList(new JdbcSourceSplit("1", "select 1", null, null)),
                     Arrays.asList(
                             new JdbcSourceSplit(
                                     "1",
                                     "select 1",
                                     new Serializable[] {new Integer(0)},
-                                    10,
                                     new CheckpointedOffset(0, 10))),
-                    Arrays.asList(new JdbcSourceSplit("1", "select 1", null, 0, null)),
+                    Arrays.asList(new JdbcSourceSplit("1", "select 1", null, null)),
                     null);
     private final JdbcSourceEnumeratorState mockedState = new MockedJdbcSourceEnumState(state);
     private final JdbcSourceEnumStateSerializer serializer =

flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/source/enumerator/JdbcSourceEnumeratorTest.java

@@ -93,7 +93,6 @@ private static JdbcSourceSplit createRandomSplit() {
                 String.valueOf(splitId++),
                 "select 1",
                 new Serializable[] {0},
-                0,
                 new CheckpointedOffset(0, 0));
     }
 

flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/source/reader/JdbcSourceReaderTest.java

@@ -51,7 +51,7 @@ void testNoSplitRequestWhenSplitRestored() throws Exception {
         final TestingReaderContext context = new TestingReaderContext();
         final JdbcSourceReader<String> reader = createReader(context);
         reader.addSplits(
-                Collections.singletonList(new JdbcSourceSplit("1", "select 1", null, 0, null)));
+                Collections.singletonList(new JdbcSourceSplit("1", "select 1", null, null)));
         reader.start();
         reader.close();
         assertThat(context.getNumSplitRequests()).isEqualTo(0);

flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/source/reader/JdbcSourceSplitReaderTest.java

@@ -49,7 +49,7 @@ class JdbcSourceSplitReaderTest extends JdbcDataTestBase {
 
     private final JdbcSourceSplit split =
             new JdbcSourceSplit(
-                    "1", "select id, title, author, price, qty from " + INPUT_TABLE, null, 0, null);
+                    "1", "select id, title, author, price, qty from " + INPUT_TABLE, null, null);
     private final JdbcConnectionProvider connectionProvider =
             new SimpleJdbcConnectionProvider(
                     new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()

flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/source/split/JdbcSourceSplitSerializerTest.java

@@ -30,7 +30,7 @@
 /** Test for {@link JdbcSourceSplitSerializer}. */
 class JdbcSourceSplitSerializerTest {
 
-    private final JdbcSourceSplit split = new JdbcSourceSplit("1", "select 1", null, 0, null);
+    private final JdbcSourceSplit split = new JdbcSourceSplit("1", "select 1", null, null);
     private final JdbcSourceSplit mockedSplit = new MockedJdbcSourceSplit(split);
     private final JdbcSourceSplitSerializer serializer = new JdbcSourceSplitSerializer();
     private final JdbcSourceSplitSerializer mockedSerializer =
@@ -73,7 +73,6 @@ public MockedJdbcSourceSplit(JdbcSourceSplit jdbcSourceSplit) {
                     jdbcSourceSplit.splitId(),
                     jdbcSourceSplit.getSqlTemplate(),
                     (Serializable[]) jdbcSourceSplit.getParameters(),
-                    jdbcSourceSplit.getOffset(),
                     jdbcSourceSplit.getCheckpointedOffset());
         }
     }