[SNOW-943288] Do not skip records when we're expecting the offset to be reset #729

sfc-gh-rcheng · 2023-10-18T00:56:34Z

https://snowflakecomputing.atlassian.net/browse/SNOW-943288

Do not skip files when there is no kafka offset reset, remove the processedOffset check

Context

Customer noticed first 500 rows are missing when using schema evolution. This is because we do not reset the kafka offset correctly when we fail in the middle of a KC batch.

Current Behavior, with buffer flush size 500:

KC sends 36 records (offset 0-35), inserted into buffer
KC sends 500 records (offset 36-535)
a. 464 (offset 36-500) records inserted into buffer and triggers flush, schema alter column, reopen channel and reset kafka offset
b. Remaining 36 records (offset 500-535) inserted into buffer <- These records should be skipped
KC resends initial 536 records, however the processedOffset has already been set to 535 due to the last 36 records (offset 500-535)

So this PR alters the logic to skip the remaining 36 records (offset 500-535)

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

sfc-gh-rcheng · 2023-10-18T18:59:09Z

test/test_suit/test_schema_evolution_w_auto_table_creation_avro_sr.py

+
+            # send second batch that should flush
+            value = []
+            for _ in range(self.flushRecordCount):


confirmed that these tests repro the issue without the changes

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

test/test_suit/test_schema_evolution_w_auto_table_creation_json.py

sfc-gh-japatel · 2023-10-25T19:46:59Z

I will take a look but can we postpone merging this fix until we have a new release?

sfc-gh-rcheng · 2023-10-31T18:36:59Z

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

@@ -488,7 +488,8 @@ public boolean hasSchemaEvolutionPermission(String tableName, String role) {
  public void appendColumnsToTable(String tableName, Map<String, String> columnToType) {
    checkConnection();
    InternalUtils.assertNotEmpty("tableName", tableName);
-    StringBuilder appendColumnQuery = new StringBuilder("alter table identifier(?) add column if not exists ");
+    StringBuilder appendColumnQuery =


required for formatter to pass for some reason. not relevant to this PR, just a small format change

sfc-gh-rcheng · 2023-11-01T16:48:06Z

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

-    // Don't skip rows if there is no offset reset or there is no offset token information in the
-    // channel
-    if (!isOffsetResetInKafka
-        || currentProcessedOffset == NO_OFFSET_TOKEN_REGISTERED_IN_SNOWFLAKE) {


crux of the change is removing this OR

sfc-gh-tzhang

Left some comments, otherwise LGTM!

test/test_suites.py

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

sfc-gh-tzhang · 2023-11-07T19:39:45Z

This is a behavior change, should we follow the new BCR release process for drivers/SDKs? cc: @sfc-gh-xhuang

sfc-gh-japatel · 2023-11-07T22:10:26Z

test/test_suit/test_schema_evolution_w_random_row_count.py

            raise NonRetryableError("Column {} was not created".format(columnName))

        res = self.driver.snowflake_conn.cursor().execute(
            "SELECT count(*) FROM {}".format(self.table)).fetchone()[0]
-        if res != self.recordNum * len(self.topics):
-            print("Number of record expected: {}, got: {}".format(self.recordNum * len(self.topics), res))
+        if res != len(self.topics) * self.recordNum:


great test, thanks!

sfc-gh-japatel

lgtm!
Thanks for the work.. Talked offline and I think this is a good short term solution. Please see if we can re-add those tests as Toby mentioned. For long term:

We do need to remove buffer in KC.
if we cant remove bufffer from KC, we should have an interface between KC buffer and the logic which returns the insertRowsResponse. Here we can discard the batch only after the bad row and not necessarily entire batch. Of course this needs to be thought through since we are talking about discarding the incoming batch from Kafka which spans across multiple partitions. (but you can route those records to individual partitions, and discard partition batches) but I do feel this is a good short term solution not relying on removal of buffer..

sfc-gh-xhuang · 2023-11-07T22:33:51Z

Let's discuss more on whether this is a behavior change or a bug fix but following the BCR process for drivers to simple enough too.

…be reset (snowflakedb#729)

sfc-gh-rcheng added 5 commits October 17, 2023 17:27

e2e test passes

8eb8036

add avro, failing though

9de5460

and not or

53db470

manual formatting

d61d160

personal nit

6996af5

sfc-gh-rcheng commented Oct 18, 2023

View reviewed changes

src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java Outdated Show resolved Hide resolved

sfc-gh-rcheng commented Oct 18, 2023

View reviewed changes

src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java Outdated Show resolved Hide resolved

sfc-gh-rcheng commented Oct 18, 2023

View reviewed changes

src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java Outdated Show resolved Hide resolved

sfc-gh-rcheng commented Oct 18, 2023

View reviewed changes

sfc-gh-rcheng changed the title ~~fix offset issue~~ [SNOW-943288] Alter logic to skip records when we're expecting the offset to be reset Oct 18, 2023

sfc-gh-rcheng added 5 commits October 19, 2023 17:21

fix tests to actually repro

09b2d62

remove drop table tests

8639534

delete drop table tests and revert insertrecordtobuffer

f84e0bc

autoformatting

fbd1df2

merge master

5188952

sfc-gh-tzhang reviewed Oct 25, 2023

View reviewed changes

add random row count and small pr comments

669e182

sfc-gh-rcheng marked this pull request as ready for review October 25, 2023 19:44

sfc-gh-rcheng requested review from sfc-gh-japatel, sfc-gh-tjones and a team as code owners October 25, 2023 19:44

sfc-gh-rcheng requested a review from sfc-gh-tzhang October 25, 2023 19:44

sfc-gh-rcheng requested a review from sfc-gh-asen October 26, 2023 17:43

sfc-gh-rcheng added 4 commits October 26, 2023 11:50

merge master

5274275

edit comment

8fe850a

remove droptable from testsuites

abc3121

Merge branch 'master' into rcheng-offsetissue

e0ea3f0

sfc-gh-rcheng commented Oct 31, 2023

View reviewed changes

sfc-gh-rcheng changed the title ~~[SNOW-943288] Alter logic to skip records when we're expecting the offset to be reset~~ [SNOW-943288] Do not skip records when we're expecting the offset to be reset Nov 1, 2023

sfc-gh-rcheng force-pushed the rcheng-offsetissue branch from b254308 to e0ea3f0 Compare November 1, 2023 16:47

sfc-gh-rcheng commented Nov 1, 2023

View reviewed changes

sfc-gh-rcheng added 3 commits November 1, 2023 09:50

remove sleeps

bde0233

autoformatting

6583b55

merge main

645d703

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

View reviewed changes

test/test_suites.py Show resolved Hide resolved

test/test_suites.py Show resolved Hide resolved

src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java Outdated Show resolved Hide resolved

sfc-gh-japatel reviewed Nov 7, 2023

View reviewed changes

sfc-gh-japatel approved these changes Nov 7, 2023

View reviewed changes

sfc-gh-rcheng added 4 commits November 7, 2023 14:35

add it

604ae6b

autoformatting

152b28a

add test back disabled

9e32250

autoformatting

14d9de3

sfc-gh-rcheng merged commit c9a3b2c into master Nov 9, 2023
32 checks passed

sfc-gh-rcheng deleted the rcheng-offsetissue branch November 9, 2023 01:55

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

[SNOW-943288] Do not skip records when we're expecting the offset to …

4181ac5

…be reset (snowflakedb#729)

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

[SNOW-943288] Do not skip records when we're expecting the offset to …

39c2c66

…be reset (snowflakedb#729)

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

[SNOW-943288] Do not skip records when we're expecting the offset to …

985b076

…be reset (snowflakedb#729)

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

[SNOW-943288] Do not skip records when we're expecting the offset to …

855c401

…be reset (snowflakedb#729)

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

[SNOW-943288] Do not skip records when we're expecting the offset to …

ce4bab7

…be reset (snowflakedb#729)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SNOW-943288] Do not skip records when we're expecting the offset to be reset #729

[SNOW-943288] Do not skip records when we're expecting the offset to be reset #729

sfc-gh-rcheng commented Oct 18, 2023 •

edited

Loading

sfc-gh-rcheng Oct 18, 2023

sfc-gh-japatel commented Oct 25, 2023

sfc-gh-rcheng Oct 31, 2023

sfc-gh-rcheng Nov 1, 2023

sfc-gh-tzhang left a comment

sfc-gh-tzhang commented Nov 7, 2023

sfc-gh-japatel Nov 7, 2023

sfc-gh-japatel left a comment

sfc-gh-xhuang commented Nov 7, 2023

[SNOW-943288] Do not skip records when we're expecting the offset to be reset #729

[SNOW-943288] Do not skip records when we're expecting the offset to be reset #729

Conversation

sfc-gh-rcheng commented Oct 18, 2023 • edited Loading

Context

sfc-gh-rcheng Oct 18, 2023

Choose a reason for hiding this comment

sfc-gh-japatel commented Oct 25, 2023

sfc-gh-rcheng Oct 31, 2023

Choose a reason for hiding this comment

sfc-gh-rcheng Nov 1, 2023

Choose a reason for hiding this comment

sfc-gh-tzhang left a comment

Choose a reason for hiding this comment

sfc-gh-tzhang commented Nov 7, 2023

sfc-gh-japatel Nov 7, 2023

Choose a reason for hiding this comment

sfc-gh-japatel left a comment

Choose a reason for hiding this comment

sfc-gh-xhuang commented Nov 7, 2023

sfc-gh-rcheng commented Oct 18, 2023 •

edited

Loading