PROD-39429 Implement migrate sys func from new channel(Format V2) to old channel (V1) - Push to Main #751

sfc-gh-japatel · 2023-11-16T23:39:50Z

PR #750 but pushed in main.

Copying it as is.

This is a long term implementation for potential data duplication introduced because of a new channel name format.

It is about moving away for customers who might have onboarded to new channel format which was introduced in version 2.1.0 (Now de-listed)
We will stick to old format for new customers running this for the first time.
Renamed channelName to channelNameFormatV1 and introduced channelNameFormatV2

Added tests in TopicPartitionChannel and TopicPartitionChannel

Notes:

This PR is opened against japatel-SNOW-971412-release-2-1-1-ready which is going to be the release branch for v2.1.1
A similar PR will also be opened against master.

End to End tests:

Use 2.1.0
Use two connectors:
Uses channel Name v2
Stop
Replace jar (2.1.1) and restart.
Does the migration and you can only see old channel format
Stop again and restart in 2.1.1
Nothing happens and we do get a valid response saying newchannelFormatV2 doesnt exist

sfc-gh-tzhang · 2023-11-18T00:20:35Z

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java

+      }
+      if (migrateOffsetTokenResultFromSysFunc == null) {
+        LOGGER.warn(
+            "No result found in Migrating OffsetToken through System Function for tableName:{},"


what does it mean by no result found? No destination channel or source channel found? or both?

If this is not expected, should we throw an exception?

This is not expected and I thought we decided to not throw any exceptions/swallow all and continue using old channel.

I can see the concern though but at this point I feel it is about whether we want to halt ingestion or atleast ignore the exception and continue moving forward with old. Halting ingestion could be better and it helps us know something is wrong rather than continuing with old channel with ramifications we dont.
WDYT Toby?

I agree that its better to swallow this exception here, but we should have something to track an error here - maybe an alert on the server side if we don't return an offset on the func call, or a new telemetry event similar to reportKafkaFatalError (if we aren't expecting too many hits on this error)

good point, both..
I think we should throw an exception if this is not expected.. I can add a server side incident and also add report telemetry..

For an unexpected exception, I am throwing a runtime exception. PTAL! It's better to fail here IMO. Thanks folks!

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java

sfc-gh-tzhang · 2023-11-18T00:22:24Z

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java

+          "Migrating OffsetToken for a SourceChannel:{} in table:{} failed due to:{}",
+          sourceChannelName,
+          fullyQualifiedTableName,
+          e.getMessage());


getMessage might be NULL for some exceptions, could we do better?

might be worth logging stacktrace too. let me add that

src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java

sfc-gh-tzhang · 2023-11-18T00:37:53Z

src/test/java/com/snowflake/kafka/connector/SnowflakeSinkTaskStreamingTest.java

@@ -88,17 +88,17 @@ public void testSinkTaskInvalidRecord_InMemoryDLQ() throws Exception {
        new TopicPartitionChannel(
            mockStreamingClient,
            topicPartition,
-            SnowflakeSinkServiceV2.partitionChannelKey(TEST_CONNECTOR_NAME, topicName, partition),


Do we have any tests that can actually test the end2end migration?

all end to end tests will eventually go through this path right? since it is default to true.

it would be hard to test this scenario in e2e since we wouldn't be able to upgrade the version. Could we add an IT (maybe in SnowflakeSinkServiceV2IT) that explicitly opens and ingests to a channel with V2 name, shuts it down, then creates a new sink and runs through the channel migration?

Not 100% sure this is possible, but maybe we can stick a mock stub for the client to hard code the channel name during open?

good idea to ingest through topicpartition channel through v2 and then let it go through migration.. I will add IT tests..

I think I see the issue, the IT tests are showing up other PR: https://github.com/snowflakedb/snowflake-kafka-connector/pull/750/files
I might have messed up with merge conflicts.

ported the IT tests here, added few more tests in connectionIT tests.

sfc-gh-rcheng

overall lgtm, but i would like an IT or e2e test that explictly opens and ingests to a V2 channel and confirms that is migrated later on

sfc-gh-rcheng · 2023-11-20T17:51:00Z

src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnectorConfig.java

+  public static final boolean ENABLE_CHANNEL_OFFSET_TOKEN_MIGRATION_DEFAULT = true;
+  public static final String ENABLE_CHANNEL_OFFSET_TOKEN_MIGRATION_DOC =
+      "This config is used to enable/disable streaming channel offset migration logic. If true, we"
+          + " will migrate offset token from channel name format V2 to name format v1.";


Nit: lets add something about how V2 is deprecated. Otherwise customers may disable this because V2 sounds fancier than V1

good point, let me add!

sfc-gh-rcheng · 2023-11-20T17:54:43Z

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java

+      }
+      if (migrateOffsetTokenResultFromSysFunc == null) {
+        LOGGER.warn(
+            "No result found in Migrating OffsetToken through System Function for tableName:{},"


I agree that its better to swallow this exception here, but we should have something to track an error here - maybe an alert on the server side if we don't return an offset on the func call, or a new telemetry event similar to reportKafkaFatalError (if we aren't expecting too many hits on this error)

...a/com/snowflake/kafka/connector/internal/streaming/ChannelMigrateOffsetTokenResponseDTO.java

sfc-gh-rcheng · 2023-11-20T18:07:31Z

src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java

@@ -278,6 +285,14 @@ public TopicPartitionChannel(

    this.enableSchemaEvolution = this.enableSchematization && hasSchemaEvolutionPermission;

+    this.channelNameFormatV2 =


lets move this channelNameFormatV2 into the if statement? I don't think V2 is needed after the channel is migrated

it was private final so had to put it outside in ctor. Let me think if this is needed as instance variable. (i had plans to modify it in tests but havent added any IT tests)

sfc-gh-rcheng · 2023-11-20T18:11:02Z

src/test/java/com/snowflake/kafka/connector/SnowflakeSinkTaskStreamingTest.java

@@ -88,17 +88,17 @@ public void testSinkTaskInvalidRecord_InMemoryDLQ() throws Exception {
        new TopicPartitionChannel(
            mockStreamingClient,
            topicPartition,
-            SnowflakeSinkServiceV2.partitionChannelKey(TEST_CONNECTOR_NAME, topicName, partition),


it would be hard to test this scenario in e2e since we wouldn't be able to upgrade the version. Could we add an IT (maybe in SnowflakeSinkServiceV2IT) that explicitly opens and ingests to a channel with V2 name, shuts it down, then creates a new sink and runs through the channel migration?

Not 100% sure this is possible, but maybe we can stick a mock stub for the client to hard code the channel name during open?

…serviceIT

sfc-gh-japatel · 2023-11-20T23:35:46Z

overall lgtm, but i would like an IT or e2e test that explictly opens and ingests to a V2 channel and confirms that is migrated later on

Sorry, the merge conflicts didnt show IT tests.
I ported IT tests from other PR, added few more connectionserviceIT tests.
e2e test involving two jars is going to be a bit difficult with our current infra, but let me think more on that.

sfc-gh-tzhang

Left some comments, PTAL, otherwise LGTM!

sfc-gh-tzhang · 2023-11-21T01:05:24Z

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java

+      LOGGER.info(
+          "Migrate OffsetToken response for table:{}, sourceChannel:{}, destinationChannel:{}"
+              + " is:{}",
+          tableName,
+          sourceChannelName,
+          destinationChannelName,
+          channelMigrateOffsetTokenResponseDTO);


This log might be confusing to customer, could we only log when there is actually a migration being done? Or do we even need this client side log given that we have a bunch of server side logging in place?

This should be fine, this is only once per channel during open partition, not very often. I would like to keep this which helps us in debugging any customer issue.

...main/java/com/snowflake/kafka/connector/internal/streaming/ChannelMigrationResponseCode.java

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java

...main/java/com/snowflake/kafka/connector/internal/streaming/ChannelMigrationResponseCode.java

…ion system function

sfc-gh-rcheng

Thanks for the change! The ITs are pretty cool - but small request to try ingestion after the migration as well

sfc-gh-rcheng · 2023-11-21T19:08:16Z

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java

+
+      String migrateOffsetTokenResultFromSysFunc = null;
+      if (resultSet.next()) {
+        migrateOffsetTokenResultFromSysFunc = resultSet.getString(1 /*Only one column*/);


out of scope for this PR - do we have any guarantees that this DTO won't change on the server side? a comment, test or something server side to warn against changes since we now throw the exception on failure

yeah, i think the json exception will help with that.. I think I have the comment on server side saying be cautious to make changes in the interface. but let me try to mimick and see what happens if response is changed. I could probably handle new fields in the object mapper.

moved the method to its own method so that we can test it. If the response changes from server side which shouldnt, this tests methods will start failing

src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java

src/test/java/com/snowflake/kafka/connector/internal/ConnectionServiceIT.java

src/test/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannelIT.java

…old channel (V1) - Push to Main (snowflakedb#751)

sfc-gh-japatel added 3 commits November 16, 2023 15:27

Remove connector name from partition channel key

cc968f3

Added config and more tests

a14b7da

Add javadocs, revert to use profile.json

c1cbaa0

sfc-gh-japatel marked this pull request as ready for review November 16, 2023 23:39

sfc-gh-japatel requested review from sfc-gh-tzhang, sfc-gh-tjones, sfc-gh-rcheng and a team as code owners November 16, 2023 23:39

Rebase fixes with main

9c6783b

sfc-gh-japatel changed the title ~~PROD-39429 migrate sys func push main~~ PROD-39429 Implement migrate sys func from new channel(Format V2) to old channel (V1) - Push to Main Nov 16, 2023

sfc-gh-tzhang reviewed Nov 18, 2023

View reviewed changes

sfc-gh-rcheng reviewed Nov 20, 2023

View reviewed changes

sfc-gh-japatel added 4 commits November 20, 2023 12:56

Addressed Toby and Revi's comments, added a response code

62100c1

Modified the return type of interface and added IT test in connection…

612027b

…serviceIT

Port IT tests from other PR

5962624

Remove instance variable

b7b5ab5

sfc-gh-japatel requested review from sfc-gh-tzhang and sfc-gh-rcheng November 20, 2023 23:33

sfc-gh-tzhang approved these changes Nov 21, 2023

View reviewed changes

Toby's suggestion about checking success before returning from migrat…

3dbf6c1

…ion system function

sfc-gh-rcheng approved these changes Nov 21, 2023

View reviewed changes

Revi's minor suggestion, more tests

21697ad

sfc-gh-japatel force-pushed the japatel-PROD-39429-migrate-sys-func-push-main branch from 0b2b4a6 to b9b1754 Compare November 21, 2023 22:51

Add end to end test which disables the migration code path

e54b031

sfc-gh-japatel force-pushed the japatel-PROD-39429-migrate-sys-func-push-main branch from b9b1754 to e54b031 Compare November 21, 2023 23:22

sfc-gh-japatel closed this Nov 22, 2023

sfc-gh-japatel reopened this Nov 22, 2023

sfc-gh-japatel merged commit 6edd211 into master Nov 22, 2023
59 of 60 checks passed

sfc-gh-japatel deleted the japatel-PROD-39429-migrate-sys-func-push-main branch November 22, 2023 05:03

khsoneji pushed a commit to confluentinc/snowflake-kafka-connector that referenced this pull request Dec 4, 2023

PROD-39429 Implement migrate sys func from new channel(Format V2) to …

b8f62c3

…old channel (V1) - Push to Main (snowflakedb#751)

khsoneji pushed a commit to confluentinc/snowflake-kafka-connector that referenced this pull request Dec 4, 2023

PROD-39429 Implement migrate sys func from new channel(Format V2) to …

8c486fe

…old channel (V1) - Push to Main (snowflakedb#751)

khsoneji pushed a commit to confluentinc/snowflake-kafka-connector that referenced this pull request Dec 4, 2023

PROD-39429 Implement migrate sys func from new channel(Format V2) to …

24724c0

…old channel (V1) - Push to Main (snowflakedb#751)

EduardHantig pushed a commit to streamkap-com/snowflake-kafka-connector that referenced this pull request Feb 1, 2024

PROD-39429 Implement migrate sys func from new channel(Format V2) to …

fe82cdd

…old channel (V1) - Push to Main (snowflakedb#751)

sudeshwasnik pushed a commit to confluentinc/snowflake-kafka-connector that referenced this pull request Feb 16, 2024

PROD-39429 Implement migrate sys func from new channel(Format V2) to …

d4d9553

…old channel (V1) - Push to Main (snowflakedb#751)

BDeus mentioned this pull request Mar 18, 2024

Rollback on Prefix with channel name with connector name #810

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

PROD-39429 Implement migrate sys func from new channel(Format V2) to old channel (V1) - Push to Main #751

PROD-39429 Implement migrate sys func from new channel(Format V2) to old channel (V1) - Push to Main #751

sfc-gh-japatel commented Nov 16, 2023 •

edited

Loading

sfc-gh-tzhang Nov 18, 2023

sfc-gh-tzhang Nov 18, 2023

sfc-gh-japatel Nov 18, 2023

sfc-gh-rcheng Nov 20, 2023

sfc-gh-japatel Nov 20, 2023

sfc-gh-japatel Nov 20, 2023

sfc-gh-tzhang Nov 18, 2023

sfc-gh-japatel Nov 18, 2023

sfc-gh-japatel Nov 20, 2023

sfc-gh-tzhang Nov 18, 2023

sfc-gh-japatel Nov 18, 2023

sfc-gh-rcheng Nov 20, 2023

sfc-gh-japatel Nov 20, 2023

sfc-gh-japatel Nov 20, 2023

sfc-gh-japatel Nov 20, 2023

sfc-gh-rcheng left a comment

sfc-gh-rcheng Nov 20, 2023

sfc-gh-japatel Nov 20, 2023

sfc-gh-rcheng Nov 20, 2023

sfc-gh-rcheng Nov 20, 2023

sfc-gh-japatel Nov 20, 2023

sfc-gh-japatel Nov 20, 2023

sfc-gh-rcheng Nov 20, 2023

sfc-gh-japatel commented Nov 20, 2023

sfc-gh-tzhang left a comment

sfc-gh-tzhang Nov 21, 2023

sfc-gh-japatel Nov 21, 2023

sfc-gh-rcheng left a comment

sfc-gh-rcheng Nov 21, 2023

sfc-gh-japatel Nov 21, 2023

sfc-gh-japatel Nov 21, 2023

		@@ -278,6 +285,14 @@ public TopicPartitionChannel(

		this.enableSchemaEvolution = this.enableSchematization && hasSchemaEvolutionPermission;

		this.channelNameFormatV2 =

PROD-39429 Implement migrate sys func from new channel(Format V2) to old channel (V1) - Push to Main #751

PROD-39429 Implement migrate sys func from new channel(Format V2) to old channel (V1) - Push to Main #751

Conversation

sfc-gh-japatel commented Nov 16, 2023 • edited Loading

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

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

sfc-gh-rcheng 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

sfc-gh-japatel commented Nov 20, 2023

sfc-gh-tzhang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sfc-gh-rcheng 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

sfc-gh-japatel commented Nov 16, 2023 •

edited

Loading