SNOW-850959: Fix a wrong result issue that offsets are skipped when schematization is enabled #658

sfc-gh-tzhang · 2023-06-22T00:07:00Z

Looks like we have a gap in KC that may skip ingesting some offsets, consider this case where you have two topics with different schemas trying to ingest into the same table, internally KC will create two channels (channel A and channel B) with offset_token=NULL, then both channels start to buffer data and flush files, but channel A fails committing the first batch because the file schema doesn't match the latest table schema due to schema evolution, then channel A will be invalidated and reopened but we won't reset the consumer offset because the offset_token for channel A is still NULL, we say that we will rely on Kafka to send us the correct data when the offset_token is NULL, so in this case Kafka will continue send us the next batch and we will accept it, this means that the first batch for channel A will be skipped forever. We need to rethink about what we need to do when the offset_token for a channel is NULL and I don't think we can purely rely on Kafka to resend us the correct offset.

The fix is to manage the Kafka consumer offset in the connector as well and use that to reset Kafka when the offset token for a channel is NULL instead of relying on Kafka to send us the correct offset

codecov · 2023-06-22T00:47:32Z

Codecov Report

Merging #658 (7686a4c) into master (52f0a8a) will decrease coverage by 0.03%.
The diff coverage is 95.45%.

@@            Coverage Diff             @@
##           master     #658      +/-   ##
==========================================
- Coverage   87.88%   87.85%   -0.03%     
==========================================
  Files          50       50              
  Lines        4144     4143       -1     
  Branches      449      451       +2     
==========================================
- Hits         3642     3640       -2     
- Misses        332      333       +1     
  Partials      170      170

Impacted Files	Coverage Δ
...ctor/internal/streaming/TopicPartitionChannel.java	`91.38% <94.73%> (-0.30%)`	⬇️
...tor/internal/streaming/SnowflakeSinkServiceV2.java	`80.86% <100.00%> (+0.23%)`	⬆️

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

sfc-gh-japatel · 2023-07-05T19:06:53Z

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

@@ -345,7 +359,8 @@ public void insertRecordToBuffer(SinkRecord kafkaSinkRecord) {
   */
  private boolean shouldIgnoreAddingRecordToBuffer(
      SinkRecord kafkaSinkRecord, long currentProcessedOffset) {
-    if (!isOffsetResetInKafka) {
+    if (!isOffsetResetInKafka
+        || currentProcessedOffset == NO_OFFSET_TOKEN_REGISTERED_IN_SNOWFLAKE) {


I am confused why is this needed. Mind adding a comment here?

We don't want to skip any rows when the offset token for a channel is NULL, think about the case when the data is expired so a row with higher offset is sent by Kafka upon restarting

sfc-gh-japatel · 2023-07-05T19:17:24Z

test/test_suites.py

+from test_suit.test_schema_evolution_avro_sr import TestSchemaEvolutionAvroSR
+from test_suit.test_schema_evolution_drop_table import TestSchemaEvolutionDropTable
+from test_suit.test_schema_evolution_json import TestSchemaEvolutionJson
+# res tests


nit: what is res?

I added that for 'resilience tests', we can add the full word there. Can we move the comment back above the rest of the resilience tests?

all the formatting is done automatically so I have no control, I simply remove it

sfc-gh-japatel · 2023-07-05T19:20:59Z

test/test_suit/test_schema_evolution_drop_table.py

@@ -46,7 +46,7 @@ def send(self):
        self.driver.sendBytesData(self.topic, value, key)

        # Sleep for some time and then verify the rows are ingested
-        sleep(60)
+        sleep(120)


Curious why it's taking more time. I see you have tried multiple times.

^ as well as why we're removing the 30 sec sleep below

good question, this solution is not perfect, if we don't wait for more than 60 seconds than the in-memory offset is not updated and we will restart from the beginning, I put 120 seconds to make sure it's not flaky

sfc-gh-japatel

lgtm! Thanks a lot!
Some minor comments to clarify.

sfc-gh-rcheng

lgtm, just want more clarification on the different offsets

sfc-gh-rcheng · 2023-07-05T19:42:53Z

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

-            this.channel.getFullyQualifiedName());
-        return latestCommittedOffsetInSnowflake;
-      }
+      LOGGER.info(


why the change here? the only diff i see is the warn log is no longer there, why not edit the content of the log instead of removing it?

I don't think warn makes sense because there are valid cases that a offset token could be NULL

We want to unify the format for both cases so it's easier to search and debug

sfc-gh-rcheng · 2023-07-05T19:45:06Z

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

@@ -266,7 +266,9 @@ public long getOffset(TopicPartition topicPartition) {
    String partitionChannelKey =
        partitionChannelKey(topicPartition.topic(), topicPartition.partition());
    if (partitionsToChannel.containsKey(partitionChannelKey)) {
-      return partitionsToChannel.get(partitionChannelKey).getOffsetSafeToCommitToKafka();
+      long offset = partitionsToChannel.get(partitionChannelKey).getOffsetSafeToCommitToKafka();


nit: can we do partitionsToChannel.get() once and save the value locally for perf?

I think this will be optimized by the complier so it's usually a matter of style?

oh good to know!

sfc-gh-rcheng · 2023-07-05T20:06:05Z

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

@@ -94,6 +94,10 @@ public class TopicPartitionChannel {
  private final AtomicLong processedOffset =
      new AtomicLong(NO_OFFSET_TOKEN_REGISTERED_IN_SNOWFLAKE);

+  // The in-memory consumer offset managed by the connector, we need this to tell Kafka which
+  // offset to resend when the channel offset token is NULL
+  private long latestConsumerOffset = NO_OFFSET_TOKEN_REGISTERED_IN_SNOWFLAKE;


the above offsetPersistedInSnowflake and processedOffset are AtomicLongs, should this also be an AtomicLong?

I believe everything should be long given that there is no concurrent logic?

sfc-gh-rcheng · 2023-07-05T20:28:40Z

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

@@ -94,6 +94,10 @@ public class TopicPartitionChannel {
  private final AtomicLong processedOffset =
      new AtomicLong(NO_OFFSET_TOKEN_REGISTERED_IN_SNOWFLAKE);

+  // The in-memory consumer offset managed by the connector, we need this to tell Kafka which


we now have three offsets, can we unify the names or clarify the different uses in the comments? From what I can tell, latestConsumerOffset is either the first KC offset (initial processedOffset) or the snowflake (offsetPersistedInSnowflake). If so, can we comment something like:

offsetPersistedInSnowflake: This offset represents the data persisted in Snowflake. More specifically it is the Snowflake offset determined from the insertRows API call. It is set after calling the fetchOffsetToken API for this channel

processedOffset: This offset represents the data buffered in KC. More specifically it is the KC offset to ensure exactly once functionality. On creation it is set to the latest committed token in Snowflake (see offsetPersistedInSnowflake) and updated on each new row from KC.

latestConsumerOffset: This offset is a fallback to represent the data buffered in KC. It is similar to processedOffset, however it is only used to resend the offset when the channel offset token is NULL. It is updated to the first offset sent by KC (see processedOffset) or the offset persisted in snowflake (see offsetPersistedInSnowflake)

thanks for the suggestion, updated

sfc-gh-rcheng · 2023-07-05T20:30:22Z

test/test_suit/test_schema_evolution_drop_table.py

@@ -46,7 +46,7 @@ def send(self):
        self.driver.sendBytesData(self.topic, value, key)

        # Sleep for some time and then verify the rows are ingested
-        sleep(60)
+        sleep(120)


^ as well as why we're removing the 30 sec sleep below

sfc-gh-rcheng · 2023-07-05T20:31:42Z

test/test_suites.py

+from test_suit.test_schema_evolution_avro_sr import TestSchemaEvolutionAvroSR
+from test_suit.test_schema_evolution_drop_table import TestSchemaEvolutionDropTable
+from test_suit.test_schema_evolution_json import TestSchemaEvolutionJson
+# res tests


I added that for 'resilience tests', we can add the full word there. Can we move the comment back above the rest of the resilience tests?

sfc-gh-rcheng · 2023-07-05T20:33:20Z

test/test_suites.py

@@ -112,91 +104,114 @@ def create_end_to_end_test_suites(driver, nameSalt, schemaRegistryAddress, testS
            test_instance=TestAvrosrAvrosr(driver, nameSalt), clean=True, run_in_confluent=True, run_in_apache=False
        )),
        ("TestNativeStringAvrosr", EndToEndTestSuite(
-            test_instance=TestNativeStringAvrosr(driver, nameSalt), clean=True, run_in_confluent=True, run_in_apache=False
+            test_instance=TestNativeStringAvrosr(driver, nameSalt), clean=True, run_in_confluent=True,


nit: why are all these 'run_in_apache' lines pushed down a line? i don't see any other differences, is there a formatting change?

all the formatting is done automatically so I have no control

…chematization is enabled (snowflakedb#658) Looks like we have a gap in KC that may skip ingesting some offsets, consider this case where you have two topics with different schemas trying to ingest into the same table, internally KC will create two channels (channel A and channel B) with offset_token=NULL, then both channels start to buffer data and flush files, but channel A fails committing the first batch because the file schema doesn't match the latest table schema due to schema evolution, then channel A will be invalidated and reopened but we won't reset the consumer offset because the offset_token for channel A is still NULL, we say that we will rely on Kafka to send us the correct data when the offset_token is NULL, so in this case Kafka will continue send us the next batch and we will accept it, this means that the first batch for channel A will be skipped forever. We need to rethink about what we need to do when the offset_token for a channel is NULL and I don't think we can purely rely on Kafka to resend us the correct offset. The fix is to manage the Kafka consumer offset in the connector as well and use that to reset Kafka when the offset token for a channel is NULL instead of relying on Kafka to send us the correct offset

sfc-gh-tzhang added 13 commits March 1, 2023 02:58

fix

f548d03

format

232150f

add tests

5eb5455

merge master

481d3f9

Merge branch 'master' into tzhang-si-fix

355f57e

Merge branch 'master' into tzhang-si-fix

db88a8e

fix perf

730ec71

fix format

778cb1b

fix tests

d5ea555

fix tests

b0bbcf9

merge master

a90f465

update behavior

d8a74a2

fix tests

8b034f4

sfc-gh-tzhang added 7 commits June 22, 2023 17:23

fix

8db222f

fix

369fc35

Merge branch 'master' into tzhang-si-fix

7ee5d21

fix

9fa5079

fix tests

5c8c098

try again

a4d2e7c

try again

c4b201e

sfc-gh-tzhang changed the title ~~[Please Ignore] Testing behavior~~ SNOW-850959: Fix a wrong result issue that offsets are skipped when schematization is enabled Jun 29, 2023

fix tests

4b884a4

sfc-gh-tzhang marked this pull request as ready for review June 29, 2023 01:16

sfc-gh-tzhang requested review from sfc-gh-japatel, sfc-gh-tjones, sfc-gh-rcheng and a team as code owners June 29, 2023 01:16

sfc-gh-tzhang added 2 commits June 28, 2023 18:20

minor change

a0308e6

try again

74465e5

sfc-gh-tzhang added 5 commits June 29, 2023 17:44

repro test failure

4c96381

repro test failure

a10a57d

try again

69d036e

try again

f38d268

try again

6c674e2

sfc-gh-japatel reviewed Jul 5, 2023

View reviewed changes

sfc-gh-japatel approved these changes Jul 5, 2023

View reviewed changes

sfc-gh-rcheng approved these changes Jul 5, 2023

View reviewed changes

sfc-gh-tzhang added 2 commits July 5, 2023 15:24

Merge branch 'master' into tzhang-si-fix

427e6cc

address comments

7686a4c

sfc-gh-tzhang merged commit 2e9c8e4 into master Jul 6, 2023
31 of 32 checks passed

sfc-gh-tzhang deleted the tzhang-si-fix branch July 6, 2023 02:07

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SNOW-850959: Fix a wrong result issue that offsets are skipped when schematization is enabled #658

SNOW-850959: Fix a wrong result issue that offsets are skipped when schematization is enabled #658

sfc-gh-tzhang commented Jun 22, 2023 •

edited

Loading

codecov bot commented Jun 22, 2023 •

edited

Loading

sfc-gh-japatel Jul 5, 2023

sfc-gh-tzhang Jul 5, 2023

sfc-gh-japatel Jul 5, 2023

sfc-gh-rcheng Jul 5, 2023

sfc-gh-tzhang Jul 5, 2023

sfc-gh-japatel Jul 5, 2023

sfc-gh-rcheng Jul 5, 2023

sfc-gh-tzhang Jul 5, 2023

sfc-gh-japatel left a comment

sfc-gh-rcheng left a comment

sfc-gh-rcheng Jul 5, 2023

sfc-gh-tzhang Jul 5, 2023

sfc-gh-rcheng Jul 5, 2023

sfc-gh-tzhang Jul 5, 2023

sfc-gh-rcheng Jul 6, 2023

sfc-gh-rcheng Jul 5, 2023

sfc-gh-tzhang Jul 5, 2023

sfc-gh-rcheng Jul 5, 2023

sfc-gh-tzhang Jul 5, 2023

sfc-gh-rcheng Jul 5, 2023

sfc-gh-rcheng Jul 5, 2023

sfc-gh-rcheng Jul 5, 2023

sfc-gh-tzhang Jul 5, 2023

SNOW-850959: Fix a wrong result issue that offsets are skipped when schematization is enabled #658

SNOW-850959: Fix a wrong result issue that offsets are skipped when schematization is enabled #658

Conversation

sfc-gh-tzhang commented Jun 22, 2023 • edited Loading

codecov bot commented Jun 22, 2023 • edited Loading

Codecov Report

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 left a 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

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-tzhang commented Jun 22, 2023 •

edited

Loading

codecov bot commented Jun 22, 2023 •

edited

Loading