CC-2116: Fix offsets compatibility #436

mageshn · 2018-06-26T14:22:33Z

With the dialect changes introduced in 5.0.x, the JDBCSourceTask always gets the table names as fully qualified table names. In the offset management, the connector was using the unqualified table name while writing the offsets and fully qualified table name for reading the offsets. Because of this when a connector restarts, it never finds the stored offsets and hence always starts from the beginning. This issue occurs only for TABLE mode.

The solution is to include a backward compatible offset management. We do the following

Introduce a new offset protocol version. This will help evolve any changes to offsets in future.
Offset writes will always follow new protocol version which is to include the FQN of the table
During read, we include both the new protocol version offset key and the old one for all tables
While selecting the offset, we always try the new protocol version key first followed by the old

This PR doesn't handle any changes to topic names. The topic names will continue to be unqualified table name to be backward compatible. If there are multiple tables, matching the intent is to error out and will be implemented in a separate PR.

(cherry picked from commit 4abd7eb)

ghost · 2018-06-26T14:22:35Z

It looks like @mageshn hasn't signed our Contributor License Agreement, yet.

The purpose of a CLA is to ensure that the guardian of a project's outputs has the necessary ownership or grants of rights over all contributions to allow them to distribute under the chosen licence.
Wikipedia

You can read and sign our full Contributor License Agreement here.

Once you've signed reply with [clabot:check] to prove it.

Appreciation of efforts,

clabot

mageshn · 2018-06-26T14:23:48Z

@rhauch I have not cleaned up the code yet and haven't added specific test yet. But wanted to get some initial feedback on the approach before spending time on those. Between, the test that was system test that was failing passes with this fix.

rhauch

A few suggestions. Biggest questions are:

why are we assigning multiple source partitions per table to the partitions map?
why not just read each source partition and offset given whatever the protocol, and then always write new partitions using the latest protocol?

rhauch · 2018-06-26T15:56:26Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

@@ -126,6 +126,7 @@ public void start(Map<String, String> properties) {
          throw new ConnectException("Unknown query mode: " + queryMode);
      }
      offsets = context.offsetStorageReader().offsets(partitions);
+      log.trace("The partition offsets are {}", offsets);


This might be nice at DEBUG level

rhauch · 2018-06-26T16:03:38Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

+            break;
+          }
+        }
+      }


This logic is looking for a single protocol amongst all of the topics, yet it is breaking as soon as it finds the first. This might be easier to understand if we put that into a separate method, with JavaDoc that explains what it is doing and the assumptions it is making.

I'm not sure that the assumption about protocol versions might be valid. Imagine a connector runs with table A and B, and offsets are both stored. Then the connector is changed to use table A only and restarted. The offsets for table A and B both use protocol V0. Then the connector is upgraded to use protocol V1, and this updates the offsets for table A. Then the connector is changed to also consume from table B, but now the offsets for table A are V1 while the offsets for table B are still V1.

Is it not sufficient to always be able to read the offsets for each partition using that partition's protocol version? Obviously all future offsets would be written with the latest protocol version.

Oh, maybe this is simply trying to find the latest protocol version to use for writes? We definitely need to extract this to a separate method and have good JavaDoc.

Why not always write using the most recent protocol version?

rhauch · 2018-06-26T16:04:32Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

+                toCheckPartition.getOrDefault(
+                    JdbcSourceConnectorConstants.OFFSET_PROTOCOL_VERSION,
+                    JdbcSourceConnectorConstants.PROTOCOL_VERSION_ZERO
+                ));


Would be good to have the TRACE log message after getting the protocol version, and then including the protocol version in the log message.

rhauch · 2018-06-26T16:08:00Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

+        JdbcSourceConnectorConstants.OFFSET_PROTOCOL_VERSION,
+        JdbcSourceConnectorConstants.PROTOCOL_VERSION_ONE)
+    ;
+    partitions.add(partitionWithFqn);


Why are we adding both the single table name and FQN as separate partitions in the map?

rhauch · 2018-06-26T16:11:22Z

src/main/java/io/confluent/connect/jdbc/source/TimestampIncrementingTableQuerier.java

+          String name = tableId.tableName(); // backward compatible
+          partition = Collections.singletonMap(JdbcSourceConnectorConstants.TABLE_NAME_KEY, name);
+          topic = topicPrefix + name;
+        }


Seems like all of this could be extracted into a strategy function to compute the topic given the table ID (e.g., topicName(TableId)), and a second method to compute the partitions for a table ID (e.g., sourcePartitions(TableId)). We could have separate strategy implementations for each protocol version, and a particular strategy implementation can be passed into this class.

rhauch · 2018-06-26T16:12:16Z

src/test/java/io/confluent/connect/jdbc/source/JdbcSourceTaskTestBase.java

+        SINGLE_TABLE_NAME);
+    SINGLE_TABLE_PARTITION_WITH_VERSION.put(
+        JdbcSourceConnectorConstants.OFFSET_PROTOCOL_VERSION,
+        "1");


This should all be moved into the PartitionProtocols class, which can have static instances of the strategy for each protocol.

style nit: the ); should be on the next line.

wicknicks · 2018-06-26T16:45:19Z

silly question: what is the problem we are solving in this PR? looks like it is related to CC-2116. Might be good to add a description above. thanks.

wicknicks

Some minor comments. Thanks, @mageshn!

wicknicks · 2018-06-26T16:44:14Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

+    partitionWithFqn.put(
+        JdbcSourceConnectorConstants.OFFSET_PROTOCOL_VERSION,
+        JdbcSourceConnectorConstants.PROTOCOL_VERSION_ONE)
+    ;


nit: the ) must be on the line with the semicolon.

wicknicks · 2018-06-26T16:49:00Z

src/test/java/io/confluent/connect/jdbc/source/JdbcSourceTaskTestBase.java

+        SINGLE_TABLE_NAME);
+    SINGLE_TABLE_PARTITION_WITH_VERSION.put(
+        JdbcSourceConnectorConstants.OFFSET_PROTOCOL_VERSION,
+        "1");


style nit: the ); should be on the next line.

wicknicks · 2018-06-26T16:52:03Z

src/test/java/io/confluent/connect/jdbc/source/JdbcSourceTaskUpdateTest.java

-    expectInitializeNoOffsets(Arrays.asList(SINGLE_TABLE_PARTITION));
+    expectInitializeNoOffsets(Arrays.asList(
+        SINGLE_TABLE_PARTITION, 
+        SINGLE_TABLE_PARTITION_WITH_VERSION));


nit: the ); should be on the next line. multiple occurrences in this file.

mageshn · 2018-06-26T20:25:42Z

@wicknicks thanks for your comments but you got too early to the PR before it was GA :)

mageshn · 2018-06-26T20:37:14Z

@rhauch I'm going to be adding a few more unit tests. But the latest changes gave me a green run for all the Mysql System tests
https://jenkins.confluent.io/job/system-test-confluent-platform-branch-builder/944/console

mageshn · 2018-06-26T21:55:15Z

@rhauch @wicknicks Do you want to take a pass at this now?

rhauch

Looks good. A few comments/suggestions.

rhauch · 2018-06-27T19:06:32Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceConnectorConstants.java

@@ -20,4 +20,6 @@
  public static final String TABLE_NAME_KEY = "table";
  public static final String QUERY_NAME_KEY = "query";
  public static final String QUERY_NAME_VALUE = "query";
+  public static final String OFFSET_PROTOCOL_VERSION_KEY = "version";


I wonder if this should be "protocol" rather than "version", since the latter might be mistaken as the Connect version.

rhauch · 2018-06-27T19:12:58Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

@@ -105,6 +109,8 @@ public void start(Map<String, String> properties) {
                                 ? Collections.singletonList(query) : tables;

    String mode = config.getString(JdbcSourceTaskConfig.MODE_CONFIG);
+    //used only in table mode
+    Map<String, List<Map<String, String>>> tableToPartitions = new HashMap<>();


I think partitionsByTableFqn would be a better name.

rhauch · 2018-06-27T19:14:59Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

-            Map<String, String> partition =
-                Collections.singletonMap(JdbcSourceConnectorConstants.TABLE_NAME_KEY, table);
-            partitions.add(partition);
+            List<Map<String, String>> tablePartitions = possibleTablePartitions(table);


I think we need a comment here that describes what we're doing:

// Find possible partition maps for different offset protocols ...

rhauch · 2018-06-27T19:17:45Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

          break;
        case QUERY:
          partition = Collections.singletonMap(
              JdbcSourceConnectorConstants.QUERY_NAME_KEY,
              JdbcSourceConnectorConstants.QUERY_NAME_VALUE
          );
+          tablePartitionsToCheck = new ArrayList<>();
+          tablePartitionsToCheck.add(partition);


Suggestion to replace these two lines with:

tablePartitionsToCheck = Collections.singleton(partition);

rhauch · 2018-06-27T19:22:49Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

-      Map<String, Object> offset = offsets == null ? null : offsets.get(partition);
+
+      //select the first matching offset
+      // This helps in handling any backward compatible offset changes


We need a better comment:

// The partition map varies by offset protocol. Since we don't know which protocol each table's offsets // are keyed by, we need to use the different possible partitions (which are in newest protocol version first) // to find the actual offsets for each table.

rhauch · 2018-06-27T19:25:55Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

+        JdbcSourceConnectorConstants.OFFSET_PROTOCOL_VERSION_KEY,
+        JdbcSourceConnectorConstants.PROTOCOL_VERSION_ONE
+    );
+


How about adding a comment:

// Define source partition map for original offset protocol (V0) that uses the table name

rhauch · 2018-06-27T19:29:14Z

src/main/java/io/confluent/connect/jdbc/source/TimestampIncrementingTableQuerier.java

+        partition.put(
+            JdbcSourceConnectorConstants.OFFSET_PROTOCOL_VERSION_KEY,
+            JdbcSourceConnectorConstants.PROTOCOL_VERSION_ONE
+        );


Didn't we talk about putting this logic into a helper method? Maybe an OffsetProtocols class with static methods:

Map<String, String> sourcePartitionForProtocolV1(TableId table) { ... } Map<String, String> sourcePartitionForProtocolV0(TableId table) { ... }

rhauch · 2018-06-27T19:30:13Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

+    Map<String, String> partition = Collections.singletonMap(
+        JdbcSourceConnectorConstants.TABLE_NAME_KEY,
+        tableId.tableName()
+    );


This should call:

Map<String, String> partition = OffsetProtocols.sourcePartitionForProtocolV0(tableId)`;

rhauch · 2018-06-27T19:30:21Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

+    partitionWithFqn.put(
+        JdbcSourceConnectorConstants.OFFSET_PROTOCOL_VERSION_KEY,
+        JdbcSourceConnectorConstants.PROTOCOL_VERSION_ONE
+    );


This should call:

Map<String, String> partition = OffsetProtocols.sourcePartitionForProtocolV1(tableId)`;

rhauch · 2018-06-27T19:32:50Z

src/test/java/io/confluent/connect/jdbc/source/JdbcSourceTaskTestBase.java

+    SINGLE_TABLE_PARTITION_WITH_VERSION.put(
+        JdbcSourceConnectorConstants.OFFSET_PROTOCOL_VERSION_KEY,
+        JdbcSourceConnectorConstants.PROTOCOL_VERSION_ONE
+    );


Could these could call OffsetProtocols methods?

mageshn · 2018-06-28T01:13:11Z

@rhauch I have incorporated the comments. Would you like to take another pass at it.

rhauch

Looks good. A few more nits.

rhauch · 2018-06-28T19:26:03Z

src/main/java/io/confluent/connect/jdbc/source/OffsetProtocols.java

+import java.util.HashMap;
+import java.util.Map;
+
+public class OffsetProtocols {


JavaDoc please?

rhauch · 2018-06-28T19:26:10Z

src/main/java/io/confluent/connect/jdbc/source/OffsetProtocols.java

+
+public class OffsetProtocols {
+
+  public static Map<String, String> sourcePartitionForProtocolV1(TableId tableId) {


JavaDoc please?

rhauch · 2018-06-28T19:26:18Z

src/main/java/io/confluent/connect/jdbc/source/OffsetProtocols.java

+    return partitionForV1;
+  }
+
+  public static Map<String, String> sourcePartitionForProtocolV0(TableId tableId) {


JavaDoc please?

rhauch · 2018-06-28T19:28:19Z

src/main/java/io/confluent/connect/jdbc/source/TimestampIncrementingTableQuerier.java

-        topic = topicPrefix + name;
+        String name = tableId.tableName();
+        topic = topicPrefix + name;// backward compatible
+        partition = OffsetProtocols.sourcePartitionForProtocolV1(tableId);


Isn't this constant for this querier instance? Can we initialize this once in the constructor and then simply reuse the same map over and over?

rhauch · 2018-06-28T19:30:47Z

src/test/java/io/confluent/connect/jdbc/source/JdbcSourceTaskUpdateTest.java

+  }
+
+  @Test
+  public void testTimestampRestore() throws Exception {


This is testing when we have offsets with both protocols. Is that worth having the method include that information? E.g., testTimestampRestoreOffsetsWithMultipleProtocol()

rhauch · 2018-06-28T19:31:33Z

src/test/java/io/confluent/connect/jdbc/source/JdbcSourceTaskUpdateTest.java

+        Collections.singletonMap(SINGLE_TABLE_PARTITION_WITH_VERSION, offset.toMap())
+    );
+  }
+
  @Test
  public void testTimestampAndIncrementingRestoreOffset() throws Exception {


This is testing when we have offsets with both protocols. Is that worth having the method include that information? E.g., testTimestampAndIncrementingRestoreOffsetWithMultipleProtocol()

rhauch · 2018-06-28T19:32:26Z

src/test/java/io/confluent/connect/jdbc/source/JdbcSourceTaskUpdateTest.java

+        Collections.singletonMap(SINGLE_TABLE_PARTITION_WITH_VERSION, offset.toMap())
+    );
+  }
+
  @Test
  public void testAutoincrementRestoreOffset() throws Exception {


This is testing when we have offsets with both protocols. Is that worth having the method include that information? E.g., testAutoincrementRestoreOffsetsWithMultipleProtocol()

rhauch · 2018-06-28T19:34:19Z

src/test/java/io/confluent/connect/jdbc/source/JdbcSourceTaskUpdateTest.java

+    Map<Map<String, String>, Map<String, Object>> offsets = new HashMap<>();
+    offsets.put(SINGLE_TABLE_PARTITION_WITH_VERSION, offset.toMap());
+    offsets.put(SINGLE_TABLE_PARTITION, oldOffset.toMap());
+    testTimestampRestoreOffset(offsets);


How about a comment here to say that we want to always use the offset with the latest protocol found?

rhauch · 2018-06-28T19:34:25Z

src/test/java/io/confluent/connect/jdbc/source/JdbcSourceTaskUpdateTest.java

+    Map<Map<String, String>, Map<String, Object>> offsets = new HashMap<>();
+    offsets.put(SINGLE_TABLE_PARTITION_WITH_VERSION, offset.toMap());
+    offsets.put(SINGLE_TABLE_PARTITION, oldOffset.toMap());
+    testTimestampAndIncrementingRestoreOffset(offsets);


How about a comment here to say that we want to always use the offset with the latest protocol found?

rhauch · 2018-06-28T19:34:32Z

src/test/java/io/confluent/connect/jdbc/source/JdbcSourceTaskUpdateTest.java

+    Map<Map<String, String>, Map<String, Object>> offsets = new HashMap<>();
+    offsets.put(SINGLE_TABLE_PARTITION_WITH_VERSION, offset.toMap());
+    offsets.put(SINGLE_TABLE_PARTITION, oldOffset.toMap());
+    testAutoincrementRestoreOffset(offsets);


How about a comment here to say that we want to always use the offset with the latest protocol found?

rhauch

LGTM. Thanks for the fix!

mageshn added 4 commits June 25, 2018 15:54

fix for offset format to be backwards compatible

27ea46c

fix for offset format to be backwards compatible

919599c

logs to debug st

f290f3f

(cherry picked from commit 4abd7eb)

Fix the partition offset compatibility

b87191e

mageshn requested a review from rhauch June 26, 2018 14:22

Fix the partition offset compatibility

ede90f6

rhauch reviewed Jun 26, 2018

View reviewed changes

wicknicks reviewed Jun 26, 2018

View reviewed changes

mageshn added 2 commits June 26, 2018 11:49

Fix the partition offset compatibility

d4aeddd

Fix formatting comments

59b75c6

Fix formatting comments

bd3e97d

mageshn added 3 commits June 26, 2018 14:36

Add unit tests to cover restore offsets scenario

0fe9cbb

Remove unused constant

31bb6c7

cleanup imports and some formatting

41b99d5

mageshn requested a review from a team June 26, 2018 21:55

rhauch changed the title ~~Fix offsets compatibility~~ CC-2250: Fix offsets compatibility Jun 27, 2018

rhauch reviewed Jun 27, 2018

View reviewed changes

Fix review comments

2f66bc8

Fix review comments

a61a819

rhauch reviewed Jun 28, 2018

View reviewed changes

Fix review comments

f23d2f8

rhauch approved these changes Jun 28, 2018

View reviewed changes

mageshn changed the title ~~CC-2250: Fix offsets compatibility~~ CC-2116: Fix offsets compatibility Jun 28, 2018

mageshn merged commit a0cd12f into 5.0.x Jun 28, 2018

mageshn deleted the CC-2116 branch June 28, 2018 20:44


		public class OffsetProtocols {

		public static Map<String, String> sourcePartitionForProtocolV1(TableId tableId) {

CC-2116: Fix offsets compatibility #436

CC-2116: Fix offsets compatibility #436

Conversation

mageshn commented Jun 26, 2018 • edited Loading

ghost commented Jun 26, 2018

mageshn commented Jun 26, 2018

rhauch left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wicknicks commented Jun 26, 2018

wicknicks left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mageshn commented Jun 26, 2018

mageshn commented Jun 26, 2018

mageshn commented Jun 26, 2018 • edited Loading

rhauch left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mageshn commented Jun 28, 2018

rhauch left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rhauch left a comment

Choose a reason for hiding this comment

mageshn commented Jun 26, 2018 •

edited

Loading

mageshn commented Jun 26, 2018 •

edited

Loading