[SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming #688

sfc-gh-rcheng · 2023-08-18T22:28:37Z

Allows tombstone records (records with NULL value and value schema) to be ingested.

Main change in RecordService creates an empty SnowflakeRecordValue when given a null value or schema

Motivation:

6 customers across slack requests and github issues
1 pager

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

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

src/main/java/com/snowflake/kafka/connector/records/RecordService.java

sfc-gh-rcheng · 2023-08-22T00:10:56Z

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

@@ -146,173 +148,6 @@ public void testIngestion() throws Exception {
    //    assert !conn.pipeExist(pipe);
  }

-  @Test
-  public void testTombstoneRecords_DEFAULT_behavior_ingestion_SFJsonConverter() throws Exception {


moved to TombstoneRecordIngestionIT.java

src/main/java/com/snowflake/kafka/connector/records/RecordService.java

sfc-gh-tzhang

LGTM, only comment is about a potential risky change which I'm not sure if it's necessary

src/main/java/com/snowflake/kafka/connector/records/RecordService.java

sfc-gh-tzhang · 2023-08-24T23:28:49Z

src/main/java/com/snowflake/kafka/connector/records/RecordService.java

@@ -297,7 +316,7 @@ public SnowflakeTableRow(SnowflakeRecordContent content, JsonNode metadata) {
  }

  void putKey(SinkRecord record, ObjectNode meta) {
-    if (record.key() == null) {
+    if (record.key() == null || record.keySchema() == null) {


why we can't put the key when the keySchema is null but not the key? This change might be risky and introduce a behavior change for both Snowpipe and Snowpipe Streaming

the current code will NPE on the next if statement if keySchema is null.

handleNativeRecord() prevents null keySchemas from being passed into putKey, however we should throw a better error if we ever get a null keySchema, so I added a null check.

This shouldn't be risky because we are changing the error thrown from NPE to a Snowflake error, what do you think?

sfc-gh-rcheng · 2023-08-24T23:39:56Z

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

@@ -749,19 +579,23 @@ public void testNativeNullIngestion() throws Exception {
    service.closeAll();
  }

-  @Test(expected = SnowflakeKafkaConnectorException.class)


We need to allow null value ingestion because community converters create null values instead of empty SnowflakeRecordContents. Currently Snowpipe fails to ingest null value records with community converters.

sfc-gh-japatel · 2023-08-25T05:21:49Z

src/main/java/com/snowflake/kafka/connector/records/RecordService.java

@@ -301,6 +322,12 @@ void putKey(SinkRecord record, ObjectNode meta) {
      return;
    }

+    if (record.keySchema() == null) {


why do we need this?

I am concerned about this change.

discussed in this comment - ensuring we throw a snowflake error instead of NPE. We should never get into this situation, but i want to add this just in case

keyschemas are often null
I dont think we should do this. this seems like a behavior change.

keyschemas are often null, we update it in handleNativeRecords earlier in the call stack. if the key schema is null in this method the next if statement throws an NPE.

The only behavior change here is to change the exception thrown from an NPE to a Snowflake Error

I See, thanks. Sounds like a harmless change but its recommended to not have refactoring/fixes in the same PR as features.

This is a small enough change to go through but it just raises more questions from reviewers POV

I see, it was caught during testing that method since i wanted to give all possible null scenarios

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

sfc-gh-japatel

lgtm

aidan-melen · 2023-08-30T15:16:15Z

What release should we expect this feature?

sfc-gh-rcheng · 2023-08-30T20:57:56Z

@aidan-melen this should go out in release 2.0.2

…snowflakedb#688)

sfc-gh-rcheng added 15 commits August 17, 2023 15:55

initial impl

f3916c5

autoformatting

eec908e

works, making it

62c1f24

autoformatting

b39a691

passes except for avro

290aad2

it passes

240d968

trying ut

a2c53e6

autoformatting

cd09713

build passes locally

928b943

personal nits

00bee04

autoformatting

1dde035

eyeball cc

62e4757

uncomment test cases

94ac2ca

autoformatting

e73a063

Merge branch 'master' into rcheng-tombstone

553999b

sfc-gh-rcheng changed the title ~~[draft] Rcheng tombstone~~ Enable tombstone record ingestion in Snowpipe Streaming Aug 21, 2023

autoformatting

754824e

sfc-gh-rcheng commented Aug 22, 2023

View reviewed changes

sfc-gh-rcheng marked this pull request as ready for review August 22, 2023 00:15

sfc-gh-rcheng requested review from sfc-gh-japatel, sfc-gh-tzhang, sfc-gh-tjones and a team as code owners August 22, 2023 00:15

sfc-gh-rcheng added 7 commits August 21, 2023 18:44

fix sinktest

6c20656

add null key failure test

b85f733

autoformatting

4ab7ffa

merge master

cafdd42

autoformatting

37dd5a7

fix dlq test

65aab68

add exception case

cfd6f54

sfc-gh-rcheng added 2 commits August 23, 2023 13:50

add null key test to streaming

631f75a

autoformatting

c815b48

sfc-gh-rcheng changed the title ~~Enable tombstone record ingestion in Snowpipe Streaming~~ [SNOW-898662] Enable tombstone record ingestion in Snowpipe Streaming Aug 23, 2023

sfc-gh-rcheng marked this pull request as ready for review August 23, 2023 20:55

sfc-gh-japatel reviewed Aug 23, 2023

View reviewed changes

src/main/java/com/snowflake/kafka/connector/records/RecordService.java Show resolved Hide resolved

update test to call set not config

eef2642

sfc-gh-rcheng changed the title ~~[SNOW-898662] Enable tombstone record ingestion in Snowpipe Streaming~~ [SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming Aug 24, 2023

group it test

06c69ba

sfc-gh-tzhang approved these changes Aug 24, 2023

View reviewed changes

sfc-gh-rcheng added 2 commits August 24, 2023 16:37

nits

abeaf39

autoformatting

3fc817f

sfc-gh-rcheng commented Aug 24, 2023

View reviewed changes

sfc-gh-rcheng added 3 commits August 24, 2023 16:58

remove null check

3fd589b

autoformatting

21cc65a

null check schema

f516110

sfc-gh-rcheng requested review from sfc-gh-tzhang and sfc-gh-japatel August 25, 2023 00:17

sfc-gh-japatel reviewed Aug 25, 2023

View reviewed changes

src/test/java/com/snowflake/kafka/connector/ConnectorConfigTest.java Show resolved Hide resolved

sfc-gh-rcheng requested a review from sfc-gh-japatel August 25, 2023 18:00

sfc-gh-japatel approved these changes Aug 25, 2023

View reviewed changes

sfc-gh-rcheng merged commit e5c0e0c into master Aug 25, 2023
30 checks passed

sfc-gh-rcheng deleted the rcheng-tombstone branch August 25, 2023 19:59

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

[SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming (…

f332205

…snowflakedb#688)

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

[SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming (…

62a780c

…snowflakedb#688)

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

[SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming (…

547c11d

…snowflakedb#688)

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

[SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming (…

cdfad49

…snowflakedb#688)

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

[SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming (…

cf2bd70

…snowflakedb#688)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming #688

[SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming #688

sfc-gh-rcheng commented Aug 18, 2023 •

edited

Loading

sfc-gh-rcheng Aug 22, 2023

sfc-gh-tzhang left a comment

sfc-gh-tzhang Aug 24, 2023

sfc-gh-rcheng Aug 25, 2023 •

edited

Loading

sfc-gh-rcheng Aug 24, 2023

sfc-gh-japatel Aug 25, 2023

sfc-gh-japatel Aug 25, 2023

sfc-gh-rcheng Aug 25, 2023 •

edited

Loading

sfc-gh-japatel Aug 25, 2023

sfc-gh-rcheng Aug 25, 2023

sfc-gh-japatel Aug 25, 2023

sfc-gh-rcheng Aug 25, 2023

sfc-gh-japatel left a comment

aidan-melen commented Aug 30, 2023

sfc-gh-rcheng commented Aug 30, 2023 •

edited

Loading

[SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming #688

[SNOW-851840] Enable tombstone record ingestion in Snowpipe Streaming #688

Conversation

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

Choose a reason for hiding this comment

sfc-gh-tzhang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sfc-gh-rcheng Aug 25, 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

sfc-gh-rcheng Aug 25, 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

sfc-gh-japatel left a comment

Choose a reason for hiding this comment

aidan-melen commented Aug 30, 2023

sfc-gh-rcheng commented Aug 30, 2023 • edited Loading

sfc-gh-rcheng commented Aug 18, 2023 •

edited

Loading

sfc-gh-rcheng Aug 25, 2023 •

edited

Loading

sfc-gh-rcheng Aug 25, 2023 •

edited

Loading

sfc-gh-rcheng commented Aug 30, 2023 •

edited

Loading