SNOW-1056407 Add SnowflakeConnectorPushTime to metadata #833

sfc-gh-lshcharbaty · 2024-05-02T13:31:41Z

Overview

SNOW-1056407
Adding a new SnowflakeConnectorPushTime property in metadata that represents a moment in time when the message was pushed further by the connector.

Depends on: #832.

Pre-review checklist

src/test/java/com/snowflake/kafka/connector/records/ProcessRecordTest.java

sfc-gh-xhuang · 2024-05-02T17:39:16Z

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

@@ -193,6 +201,10 @@ private SnowflakeTableRow processRecord(SinkRecord record) {
      meta.put(SCHEMA_ID, valueContent.getSchemaID());
    }

+    if (metadataConfig.connectorTimeFlag) {
+      meta.put(CONNECTOR_TIME, clock.instant().toEpochMilli());


could we get the timestamp of when insertRow(s) to the streaming SDK happened?

+1, I think we can do better to use the timestamp when we move from KC buffer to Ingest buffer

This method is called right before pushing data to ingest-sdk. The metadata is written as json immediately, so if we want to call it in the exact place where there rows are being pushed further, refactoring of the record service is needed.

That's the call stack:
TopicPartitionChannel.insertRowsWithFallback -> InsertRowsApiResponseSupplier -> StreamingBuffer.getData() -> RecordService.getProcessedRecordForStreamingIngest -> RecordService.processRecord.

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

sfc-gh-tzhang · 2024-05-02T22:04:16Z

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

@@ -193,6 +201,10 @@ private SnowflakeTableRow processRecord(SinkRecord record) {
      meta.put(SCHEMA_ID, valueContent.getSchemaID());
    }

+    if (metadataConfig.connectorTimeFlag) {
+      meta.put(CONNECTOR_TIME, clock.instant().toEpochMilli());


+1, I think we can do better to use the timestamp when we move from KC buffer to Ingest buffer

sfc-gh-japatel · 2024-05-07T17:58:24Z

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

@@ -69,6 +71,7 @@ public class RecordService {
  static final String CONTENT = "content";
  static final String META = "meta";
  static final String SCHEMA_ID = "schema_id";
+  static final String CONNECTOR_PUSH_TIME = "ConnectorPushTime";


we have used all lower case historically
to be consistent, may be use connector_push_time?

nvm, I see CreateTime

should we just have PushTime (or flush) time to make it consistent with CreateTime?

Imo PushTime sounds quite ambiguous. It isn't clear where exactly the push happened. Ideally I'd have SnowflakeConnectorPushTime, but it's too long and I reckon in most cases it should be obvious for users what connector is implied.

I'd like not to use "flush", as we still have an ingest-sdk that buffers data, so it isn't a time when a real flush to Snowflake happens. Also, after removing a buffering layer in the connector, nothing is going to be flushed as records will be passed to ingest-sdk immediately.

I vote for SnowflakeConnectorPushTime too.

TBH, it should be SnowflakeConnectorPullTime to be very specific. (Kafka connect polling so w.r.t snowflake it is pull time)

i am fine with either given we will and should be very descriptive in our docs.

I would even go with adding Ms in suffix

I'm avoiding pull for pull-kafka reasons.

I'm not sure if ms is necessary as the other createtime doesn;t have it and it's also millisecond value?

After reconsideration, naming the field as SnowflakeConnectorPushTime makes more sense, as customers can use some source connector, so it isn't clear which one's timestamp it is.

I wouldn't go with ms suffix in order to align with the Kafka's naming for timestamp: CreateTime, LogAppendTime.

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

sfc-gh-japatel · 2024-05-07T20:03:33Z

test/test_suit/test_avro_avro.py

@@ -31,7 +31,7 @@ def verify(self, round):
        # validate content of line 1
        res = self.driver.snowflake_conn.cursor().execute(
            "Select * from {} limit 1".format(self.topic)).fetchone()
-        goldMeta = r'{"CreateTime":\d*,"key":[{"timestamp":\d*,"tweet":"Rock:Nerfpaper,scissorsisfine.",' \
+        goldMeta = r'{"CreateTime":\d*,"SnowflakeConnectorTime":\d*,"key":[{"timestamp":\d*,"tweet":"Rock:Nerfpaper,scissorsisfine.",' \


is the default set to true? this seems like a breaking change?

+1, default should be false. Also the name doesn't seem to much with above

Turned the parameter false by default. Yes, the property name in the test had not been updated by that time.

We can have client breaking changes. It just needs to be documented.
I don't think having it default true is a bad thing

Shouldn't the connector follow BCR too?
If not then both options sound alright

Client BCRs are different than Snowflake server release BCRs. Yes, there is still approval and review but to customers, there is a notice and customers can choose to upgrade to the newest client version.
I'll forward the internal doc.

sfc-gh-japatel

lgtm overall, I will let @sfc-gh-xhuang pitch in on the name of the field.

my main concern is test changes, since default is false, why are null fields showing up in goldMeta? If they are just regex, please ignore.

It would be good to add a test for enabling/disabling the parameter to verify expected behavior.

sfc-gh-tzhang

Left some suggestions about the naming, feel free to discuss it with the team and PMs and you can make the final call
The default of this property should be false at the beginning, we can consider enable it some time later by default as a behavior change

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

sfc-gh-tzhang · 2024-05-08T05:28:10Z

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

@@ -69,6 +71,7 @@ public class RecordService {
  static final String CONTENT = "content";
  static final String META = "meta";
  static final String SCHEMA_ID = "schema_id";
+  static final String CONNECTOR_PUSH_TIME = "ConnectorPushTime";


should we just have PushTime (or flush) time to make it consistent with CreateTime?

sfc-gh-tzhang · 2024-05-08T05:34:52Z

test/test_suit/test_avro_avro.py

@@ -31,7 +31,7 @@ def verify(self, round):
        # validate content of line 1
        res = self.driver.snowflake_conn.cursor().execute(
            "Select * from {} limit 1".format(self.topic)).fetchone()
-        goldMeta = r'{"CreateTime":\d*,"key":[{"timestamp":\d*,"tweet":"Rock:Nerfpaper,scissorsisfine.",' \
+        goldMeta = r'{"CreateTime":\d*,"SnowflakeConnectorTime":\d*,"key":[{"timestamp":\d*,"tweet":"Rock:Nerfpaper,scissorsisfine.",' \


+1, default should be false. Also the name doesn't seem to much with above

sfc-gh-japatel

lgtm, left a comment.
Thanks for reverting to false by default.

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

sfc-gh-japatel · 2024-05-08T16:39:09Z

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

@@ -69,6 +71,7 @@ public class RecordService {
  static final String CONTENT = "content";
  static final String META = "meta";
  static final String SCHEMA_ID = "schema_id";
+  static final String CONNECTOR_PUSH_TIME = "ConnectorPushTime";


I would even go with adding Ms in suffix

sfc-gh-mbobowski · 2024-05-09T14:17:20Z

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

@@ -1213,7 +1214,7 @@ protected long getApproxSizeOfRecordInBytes(SinkRecord kafkaSinkRecord) {
    try {
      // get the row that we want to insert into Snowflake.
      Map<String, Object> tableRow =
-          recordService.getProcessedRecordForStreamingIngest(snowflakeRecord);
+          recordService.getProcessedRecordForStreamingIngest(snowflakeRecord, Instant.now());


What do you think about creating a TimeProvider abstraction and passing its implementation to the constructor of RecordService?

sfc-gh-mbobowski

LGTM - I would consider adding TimeProvider abstraction

sfc-gh-lshcharbaty · 2024-05-10T12:41:40Z

Since SnowflakeConnectorPushTime is included in record_metadata by default, this change is a subject to BCR

)

sfc-gh-lshcharbaty requested review from sfc-gh-japatel, sfc-gh-tzhang, sfc-gh-tjones, sfc-gh-rcheng and a team as code owners May 2, 2024 13:31

sfc-gh-lshcharbaty commented May 2, 2024

View reviewed changes

src/test/java/com/snowflake/kafka/connector/records/ProcessRecordTest.java Outdated Show resolved Hide resolved

sfc-gh-gjachimko approved these changes May 2, 2024

View reviewed changes

sfc-gh-xhuang reviewed May 2, 2024

View reviewed changes

sfc-gh-tzhang requested changes May 2, 2024

View reviewed changes

sfc-gh-lshcharbaty force-pushed the lshcharbaty/SNOW-1056407_Add_MetaTest_for_streaming branch 2 times, most recently from 321d3bd to 2997ca4 Compare May 7, 2024 07:53

sfc-gh-lshcharbaty force-pushed the lshcharbaty/SNOW-1056407_Add_connector_timestamp branch from 30028d3 to 5276143 Compare May 7, 2024 08:24

sfc-gh-lshcharbaty requested review from sfc-gh-tzhang and sfc-gh-xhuang May 7, 2024 08:24

Base automatically changed from lshcharbaty/SNOW-1056407_Add_MetaTest_for_streaming to master May 7, 2024 08:47

sfc-gh-lshcharbaty added 2 commits May 7, 2024 11:03

SNOW-1056407 Add SnowflakeConnectorTime to metadata

1a03f04

SNOW-1056407 Write ConnectorPushTime for streaming only

a7f5a4d

sfc-gh-lshcharbaty force-pushed the lshcharbaty/SNOW-1056407_Add_connector_timestamp branch from 5276143 to a7f5a4d Compare May 7, 2024 09:03

sfc-gh-lshcharbaty changed the title ~~SNOW-1056407 Add SnowflakeConnectorTime to metadata~~ SNOW-1056407 Add ConnectorPushTime to metadata May 7, 2024

sfc-gh-japatel reviewed May 7, 2024

View reviewed changes

src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnectorConfig.java Outdated Show resolved Hide resolved

sfc-gh-japatel reviewed May 7, 2024

View reviewed changes

sfc-gh-japatel approved these changes May 7, 2024

View reviewed changes

sfc-gh-tzhang approved these changes May 8, 2024

View reviewed changes

sfc-gh-lshcharbaty force-pushed the lshcharbaty/SNOW-1056407_Add_connector_timestamp branch from b0624e9 to 48a42cb Compare May 8, 2024 09:04

SNOW-1056407 Disable push time by default

eed76f7

sfc-gh-lshcharbaty force-pushed the lshcharbaty/SNOW-1056407_Add_connector_timestamp branch from 48a42cb to eed76f7 Compare May 8, 2024 09:27

sfc-gh-japatel approved these changes May 8, 2024

View reviewed changes

sfc-gh-lshcharbaty added 2 commits May 9, 2024 11:07

SNOW-1056407 Optimize imports

4dc166a

SNOW-1056407 Rename the metadata field to SnowflakeConnectorPushTime

116c403

sfc-gh-mbobowski reviewed May 9, 2024

View reviewed changes

sfc-gh-mbobowski approved these changes May 9, 2024

View reviewed changes

sfc-gh-lshcharbaty added 2 commits May 10, 2024 13:31

SNOW-1056407 Generate a timestamp in RecordService

dfeaf1b

SNOW-1056407 Enable SnowflakeConnectorPushTime by default

6a73810

sfc-gh-lshcharbaty force-pushed the lshcharbaty/SNOW-1056407_Add_connector_timestamp branch from c0a16ec to 6a73810 Compare May 10, 2024 11:55

sfc-gh-lshcharbaty changed the title ~~SNOW-1056407 Add ConnectorPushTime to metadata~~ SNOW-1056407 Add SnowflakeConnectorPushTime to metadata May 10, 2024

sfc-gh-mbobowski merged commit 45a6f8e into master May 15, 2024
81 checks passed

sfc-gh-mbobowski deleted the lshcharbaty/SNOW-1056407_Add_connector_timestamp branch May 15, 2024 08:12

chrisluedtke mentioned this pull request Jul 15, 2024

Default value expressions not supported - snowpipe streaming #668

Closed

ConfluentSemaphore pushed a commit to confluentinc/snowflake-kafka-connector that referenced this pull request Aug 6, 2024

SNOW-1056407 Add SnowflakeConnectorPushTime to metadata (snowflakedb#833

d2648ec

)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SNOW-1056407 Add SnowflakeConnectorPushTime to metadata #833

SNOW-1056407 Add SnowflakeConnectorPushTime to metadata #833

sfc-gh-lshcharbaty commented May 2, 2024 •

edited

Loading

sfc-gh-xhuang May 2, 2024

sfc-gh-tzhang May 2, 2024 •

edited

Loading

sfc-gh-lshcharbaty May 7, 2024

sfc-gh-tzhang May 2, 2024 •

edited

Loading

sfc-gh-japatel May 7, 2024

sfc-gh-japatel May 7, 2024

sfc-gh-tzhang May 8, 2024

sfc-gh-lshcharbaty May 8, 2024

sfc-gh-japatel May 8, 2024

sfc-gh-japatel May 8, 2024

sfc-gh-xhuang May 8, 2024

sfc-gh-lshcharbaty May 9, 2024

sfc-gh-japatel May 7, 2024

sfc-gh-tzhang May 8, 2024

sfc-gh-lshcharbaty May 8, 2024 •

edited

Loading

sfc-gh-xhuang May 8, 2024

sfc-gh-lshcharbaty May 9, 2024

sfc-gh-xhuang May 9, 2024

sfc-gh-japatel left a comment

sfc-gh-tzhang left a comment

sfc-gh-tzhang May 8, 2024

sfc-gh-tzhang May 8, 2024

sfc-gh-japatel left a comment

sfc-gh-japatel May 8, 2024

sfc-gh-mbobowski May 9, 2024

sfc-gh-mbobowski left a comment

sfc-gh-lshcharbaty commented May 10, 2024

SNOW-1056407 Add SnowflakeConnectorPushTime to metadata #833

SNOW-1056407 Add SnowflakeConnectorPushTime to metadata #833

Conversation

sfc-gh-lshcharbaty commented May 2, 2024 • edited Loading

Overview

Pre-review checklist

Choose a reason for hiding this comment

sfc-gh-tzhang May 2, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sfc-gh-tzhang May 2, 2024 • 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

sfc-gh-lshcharbaty May 8, 2024 • 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-japatel left a comment

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

Choose a reason for hiding this comment

sfc-gh-japatel 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-mbobowski left a comment

Choose a reason for hiding this comment

sfc-gh-lshcharbaty commented May 10, 2024

sfc-gh-lshcharbaty commented May 2, 2024 •

edited

Loading

sfc-gh-tzhang May 2, 2024 •

edited

Loading

sfc-gh-tzhang May 2, 2024 •

edited

Loading

sfc-gh-lshcharbaty May 8, 2024 •

edited

Loading