[Streaming Telemetry 1] Prepares TopicPartitionChannel for telemetry #696

sfc-gh-rcheng · 2023-08-30T23:25:30Z

Requires TelemetryService not null in TopicPartitionChannel
Adds offset setters in TopicPartitionChannel to update offset and SnowflakeTelemetryChannelStatus offsets at the same time
Moves JMX into new SnowflakeTelemetryChannelStatus

Changes coming in part 2 PR

SnowflakeTelemetryChannelStatus becomes a telemetry object
Enables reportKafkaPartitionStart/Usage telemetry

sfc-gh-rcheng · 2023-08-30T23:28:54Z

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

@@ -284,12 +291,6 @@ public TopicPartitionChannel(
              + " correct offset instead",
          this.getChannelName());
    }
-
-    if (enableCustomJMXMonitoring) {


moved into snowflakeTelemetryChannelStatus constructor

sfc-gh-rcheng · 2023-08-30T23:29:53Z

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

@@ -1095,10 +1127,9 @@ protected SnowflakeTelemetryService getTelemetryServiceV2() {
    return this.telemetryServiceV2;
  }

-  protected void setLatestConsumerOffset(long consumerOffset) {


moved up to the other offset setters

sfc-gh-rcheng · 2023-08-30T23:30:15Z

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

-   * @param metricsJmxReporter wrapper class for registering all metrics related to above connector
-   *     and channel
-   */
-  public void registerChannelJMXMetrics(


moved into SnowflakeTelemetryChannelStatus

sfc-gh-rcheng · 2023-08-30T23:30:49Z

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java

+import com.snowflake.kafka.connector.internal.metrics.MetricsUtil;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class SnowflakeTelemetryChannelStatus {


Will extend SnowflakeTelemetryBasicInfo and become a proper telemetry object in the next PR

.../com/snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryServiceV2.java

sfc-gh-rcheng · 2023-08-30T23:31:49Z

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

@@ -817,27 +835,52 @@ public void testBigAvroBufferBytesThreshold() throws Exception {
  }

  @Test
-  public void testRegisterJmxMetrics() {


moved into SnowflakeTelemetryChannelStatusTest

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

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java

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

sfc-gh-japatel

overall lgtm, couple of questions and one request to add a new test.

sfc-gh-tzhang

left some comments, PTAL, thanks!

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

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

sfc-gh-tzhang · 2023-09-02T01:35:00Z

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java

+      // offsets
+      currentMetricRegistry.register(
+          constructMetricName(
+              this.channelName,


Not familiar with JMX, but can a channel name be uniquely identify a metric? What if we have two channels with the same name but different client sequencer?

Channel names are meant to be unique identifiers for that particular topic and partition.

From what I understand, client sequencers are bumped up when a channel is reopened, which means the registry should replace the previous channel name's metrics. This should be ok behavior, wdyt?

Another potential issue is two connectors listening to the same topic, however this is not an issue because the MetricRegistry is created off the unique connector name.

what about during rebalance? Any concern about emit metrics from two nodes?

There shouldn't be rebalance issues because JmxReporters are created at the serviceSink level. When registering new metrics for a TopicPartitionChannel during rebalance, we remove all existing metrics and then register new ones.

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java

sfc-gh-rcheng · 2023-09-05T19:25:04Z

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

    mockTelemetryService = Mockito.mock(SnowflakeTelemetryService.class);
    Mockito.when(mockStreamingClient.isClosed()).thenReturn(false);
    Mockito.when(mockStreamingClient.openChannel(ArgumentMatchers.any(OpenChannelRequest.class)))
        .thenReturn(mockStreamingChannel);
    Mockito.when(mockStreamingChannel.getFullyQualifiedName()).thenReturn(TEST_CHANNEL_NAME);
    this.topicPartition = new TopicPartition(TOPIC, PARTITION);
    this.sfConnectorConfig = TestUtils.getConfig();
-    this.streamingBufferThreshold = new StreamingBufferThreshold(10, 10_000, 1);
+    this.streamingBufferThreshold = new StreamingBufferThreshold(1, 10_000, 1);


reduced the time flush to 1 second to replicate our recommended settings

sfc-gh-tzhang

LGTM, ship it!

sfc-gh-tzhang · 2023-09-07T02:56:09Z

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java

+import com.snowflake.kafka.connector.internal.metrics.MetricsUtil;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class SnowflakeTelemetryChannelStatus {


add a comment about what this class is doing?

will be added in the next PR

sfc-gh-tzhang · 2023-09-07T03:09:38Z

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java

+      // offsets
+      currentMetricRegistry.register(
+          constructMetricName(
+              this.channelName,


what about during rebalance? Any concern about emit metrics from two nodes?

…nowflakedb#696)

sfc-gh-rcheng added 18 commits August 25, 2023 15:18

initial imp

0775e19

autoformatting

273c8d7

actually send telem

340d0d0

autoformatting

ad0a854

refactoer test

ec5352e

added creation

59af12c

autoformatting

038aa7f

make telem nonnull required

ad9b223

autoformatting

f576c0f

personal nits

92bf65f

refactoring

16833c6

autoformatting

59786f0

split pr

606c9de

autoformatting

deedefa

take out telem changes

692f870

autoformatting

1f61b92

add test back

e101baf

autoformatting

170602e

sfc-gh-rcheng commented Aug 30, 2023

View reviewed changes

.../com/snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryServiceV2.java Outdated Show resolved Hide resolved

sfc-gh-rcheng commented Aug 30, 2023

View reviewed changes

personal nits

395a7b6

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

sfc-gh-rcheng requested review from sfc-gh-japatel, sfc-gh-tzhang, sfc-gh-tjones and a team as code owners August 30, 2023 23:53

sfc-gh-japatel reviewed Aug 31, 2023

View reviewed changes

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

sfc-gh-japatel reviewed Aug 31, 2023

View reviewed changes

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java Show resolved Hide resolved

sfc-gh-japatel reviewed Aug 31, 2023

View reviewed changes

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java Outdated Show resolved Hide resolved

sfc-gh-japatel reviewed Aug 31, 2023

View reviewed changes

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java Outdated Show resolved Hide resolved

sfc-gh-japatel reviewed Aug 31, 2023

View reviewed changes

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

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

View reviewed changes

sfc-gh-rcheng added 3 commits August 31, 2023 15:03

pr reviews

b58121e

autoformatting

88f19ec

pipename nits

c18e1e6

sfc-gh-rcheng requested a review from sfc-gh-japatel August 31, 2023 22:08

sfc-gh-tzhang reviewed Sep 2, 2023

View reviewed changes

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java Show resolved Hide resolved

sfc-gh-tzhang reviewed Sep 2, 2023

View reviewed changes

.../snowflake/kafka/connector/internal/streaming/telemetry/SnowflakeTelemetryChannelStatus.java Outdated Show resolved Hide resolved

sfc-gh-rcheng added 6 commits September 5, 2023 12:06

dont copy

0a129d0

autoformatting

487004f

cleanup offset copy removal

b0a6dba

autoformatting

4e1fe79

update latest consumer offset to atomic long

c224dab

autoformatting

1b68342

sfc-gh-rcheng commented Sep 5, 2023

View reviewed changes

sfc-gh-rcheng requested a review from sfc-gh-tzhang September 5, 2023 19:26

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

View reviewed changes

sfc-gh-rcheng merged commit a87e8b5 into master Sep 7, 2023

sfc-gh-rcheng deleted the rcheng-sstelem-1 branch September 7, 2023 17:52

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

[Streaming Telemetry 1] Prepares TopicPartitionChannel for telemetry (s…

285576b

…nowflakedb#696)

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

[Streaming Telemetry 1] Prepares TopicPartitionChannel for telemetry (s…

f981f1c

…nowflakedb#696)

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

[Streaming Telemetry 1] Prepares TopicPartitionChannel for telemetry (s…

ae332bb

…nowflakedb#696)

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

[Streaming Telemetry 1] Prepares TopicPartitionChannel for telemetry (s…

5d35f19

…nowflakedb#696)

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

[Streaming Telemetry 1] Prepares TopicPartitionChannel for telemetry (s…

b0b499e

…nowflakedb#696)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Streaming Telemetry 1] Prepares TopicPartitionChannel for telemetry #696

[Streaming Telemetry 1] Prepares TopicPartitionChannel for telemetry #696

sfc-gh-rcheng commented Aug 30, 2023 •

edited

Loading

sfc-gh-rcheng Aug 30, 2023

sfc-gh-rcheng Aug 30, 2023

sfc-gh-rcheng Aug 30, 2023

sfc-gh-rcheng Aug 30, 2023

sfc-gh-rcheng Aug 30, 2023

sfc-gh-japatel left a comment

sfc-gh-tzhang left a comment

sfc-gh-tzhang Sep 2, 2023

sfc-gh-rcheng Sep 5, 2023

sfc-gh-tzhang Sep 7, 2023

sfc-gh-rcheng Sep 7, 2023

sfc-gh-rcheng Sep 5, 2023

sfc-gh-tzhang left a comment

sfc-gh-tzhang Sep 7, 2023

sfc-gh-rcheng Sep 7, 2023

sfc-gh-tzhang Sep 7, 2023

[Streaming Telemetry 1] Prepares TopicPartitionChannel for telemetry #696

[Streaming Telemetry 1] Prepares TopicPartitionChannel for telemetry #696

Conversation

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this 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

Choose a reason for hiding this comment

sfc-gh-rcheng commented Aug 30, 2023 •

edited

Loading