SNOW-1061848: DirectTopicPartitionChannel - copy of the old implementation with removed buffer #852

sfc-gh-akowalczyk · 2024-05-29T11:10:29Z

Overview

SNOW-1061848

This is my second attempt to remove the buffering from a quite tightly convoluted piece of code in the BufferredTopicPartitionChannel class.

To reduce the extent of changes, and make it easily comparable to the old implementation, the PR is prepared in 3 steps (1 step is 1 commit).

DirectTopicPartionChannel is just a clone of BufferrredTopicPartitionChannel - you can skip browsing this commit.
My attempt to delete all buffers-related classes and make the code work again. Any doubts were solved by adapting Revi changes from the PR Rcheng doublebuff2 #735
Fix of unit tests to make them green

The current implementation is validated with:

TopicPartitionChannelTest - all green
some of TopicPartitionChannelIT -however it still requires manual change in the code to follow a single buffer path.

This change is safe to merge as we still do not allow for using a single buffer. As the next PRs I'd like to gradually make other tests passing.

Pre-review checklist

sfc-gh-gjachimko · 2024-05-29T12:42:39Z

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

+   * @return InsertValidationResponse a response that wraps around InsertValidationResponse
+   */
+  private InsertValidationResponse insertRowsWithFallback(Map<String, Object> transformedRecord, long offset) {
+    Fallback<Object> reopenChannelFallbackExecutorForInsertRows =


i know this code was here initially, but it does obscure the view a lot. what do you think - could we perhaps extract the failsafe handling outside, to some other class, make it sort of an aspect applied on top of the channel, so from channel perspective- we simply insert rows?

while this code is functional, its not readable or easily debuggable. Probably out of scope for this pr, but yes i support either splitting the fallback logic into a different class or using a different implementation

sfc-gh-gjachimko

i think it does the job, however, the amount of state managed by the channel is huge.
shall we consider refactoring it into smaller pieces?
i think for scenarios like that, building something like a flux stream could significantly reduce the complexity.. but that's something we could consider later.

sfc-gh-lshcharbaty

Great job!
Apply ./format.sh, please.

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

sfc-gh-lshcharbaty · 2024-05-29T13:12:50Z

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

   */
-  private InsertRowsResponse insertRowsWithFallback(StreamingBuffer buffer) {
+  private InsertValidationResponse insertRowsWithFallback(Map<String, Object> transformedRecord, long offset) {


Nit: insertRowWithFallback.

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

sfc-gh-rcheng

lgtm thanks for the change, i like how you moved the schematization logic. left comments about logging and other small changes

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

sfc-gh-rcheng · 2024-05-29T19:33:04Z

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

+      return;
+    }
+    // Accept the incoming record only if we don't have a valid offset token at server side, or the
+    // incoming record offset is 1 + the processed offset


replace 1+ with greater in the comment here to match behavior

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

sfc-gh-rcheng · 2024-05-29T19:41:36Z

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

+   * @return InsertValidationResponse a response that wraps around InsertValidationResponse
+   */
+  private InsertValidationResponse insertRowsWithFallback(Map<String, Object> transformedRecord, long offset) {
+    Fallback<Object> reopenChannelFallbackExecutorForInsertRows =


while this code is functional, its not readable or easily debuggable. Probably out of scope for this pr, but yes i support either splitting the fallback logic into a different class or using a different implementation

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

sfc-gh-rcheng · 2024-05-29T19:55:53Z

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

  }

  @Parameterized.Parameters(name = "{0}")
  public static Collection<Object[]> input() {
-    return Arrays.asList(new Object[][] {{true}, {false}});
+    return Arrays.asList(new Object[][] {


lets name the variables similar to this test for better readability in intellij

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

sfc-gh-rcheng · 2024-05-29T20:04:08Z

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

-   *     being invoked)
-   * @return Approximate long size of record in bytes. 0 if record is broken
-   */
-  protected long getApproxSizeOfRecordInBytes(SinkRecord kafkaSinkRecord) {


oh nice were removing this! we actually make this approximation twice per record, so hopefully this will speed up ingestion to not make this pass through

…ation with removed buffer (snowflakedb#852)

sfc-gh-akowalczyk requested a review from a team as a code owner May 29, 2024 11:10

sfc-gh-akowalczyk added 3 commits May 29, 2024 13:12

Create a DirectTopicPartitionChannel as a copy of the buffered one.

cfe7b64

DirectTopicPartitionChannel - removing a buffer.

3023346

Try unit tests

a4ce755

sfc-gh-akowalczyk force-pushed the akowalczyk/snow-1061848-direct-topic-channel-body branch from 2615388 to a4ce755 Compare May 29, 2024 11:12

sfc-gh-gjachimko reviewed May 29, 2024

View reviewed changes

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

View reviewed changes

sfc-gh-lshcharbaty approved these changes May 29, 2024

View reviewed changes

sfc-gh-rcheng approved these changes May 29, 2024

View reviewed changes

sfc-gh-akowalczyk added 2 commits June 3, 2024 14:02

PR comments #1

ef23b67

Formatting

906e205

sfc-gh-akowalczyk requested a review from a team as a code owner June 3, 2024 13:04

PR comments #2

0ae0d55

sfc-gh-akowalczyk force-pushed the akowalczyk/snow-1061848-direct-topic-channel-body branch from cc2b6f9 to 0ae0d55 Compare June 3, 2024 13:25

sfc-gh-akowalczyk merged commit 951f39f into master Jun 3, 2024
78 of 81 checks passed

sfc-gh-akowalczyk deleted the akowalczyk/snow-1061848-direct-topic-channel-body branch June 3, 2024 14:29

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

SNOW-1061848: DirectTopicPartitionChannel - copy of the old implement…

04b4c72

…ation with removed buffer (snowflakedb#852)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SNOW-1061848: DirectTopicPartitionChannel - copy of the old implementation with removed buffer #852

SNOW-1061848: DirectTopicPartitionChannel - copy of the old implementation with removed buffer #852

sfc-gh-akowalczyk commented May 29, 2024 •

edited

Loading

sfc-gh-gjachimko May 29, 2024

sfc-gh-rcheng May 29, 2024

sfc-gh-gjachimko left a comment

sfc-gh-lshcharbaty left a comment

sfc-gh-lshcharbaty May 29, 2024

sfc-gh-rcheng left a comment

sfc-gh-rcheng May 29, 2024

sfc-gh-rcheng May 29, 2024

sfc-gh-rcheng May 29, 2024

sfc-gh-rcheng May 29, 2024

SNOW-1061848: DirectTopicPartitionChannel - copy of the old implementation with removed buffer #852

SNOW-1061848: DirectTopicPartitionChannel - copy of the old implementation with removed buffer #852

Conversation

sfc-gh-akowalczyk commented May 29, 2024 • edited Loading

Overview

Pre-review checklist

sfc-gh-gjachimko May 29, 2024

Choose a reason for hiding this comment

sfc-gh-rcheng May 29, 2024

Choose a reason for hiding this comment

sfc-gh-gjachimko left a comment

Choose a reason for hiding this comment

sfc-gh-lshcharbaty left a comment

Choose a reason for hiding this comment

sfc-gh-lshcharbaty May 29, 2024

Choose a reason for hiding this comment

sfc-gh-rcheng left a comment

Choose a reason for hiding this comment

sfc-gh-rcheng May 29, 2024

Choose a reason for hiding this comment

sfc-gh-rcheng May 29, 2024

Choose a reason for hiding this comment

sfc-gh-rcheng May 29, 2024

Choose a reason for hiding this comment

sfc-gh-rcheng May 29, 2024

Choose a reason for hiding this comment

sfc-gh-akowalczyk commented May 29, 2024 •

edited

Loading