oops revert streaming jmx changes

snowflakedb · Jul 25, 2023 · 85dc5d3 · 85dc5d3
1 parent f496ee8
commit 85dc5d3
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 46 deletions.
diff --git a/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnectorConfig.java b/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnectorConfig.java
@@ -40,7 +40,7 @@
  */
 public class SnowflakeSinkConnectorConfig {
 
-  public static final String NAME = Utils.NAME;
+  static final String NAME = Utils.NAME;
   public static final String TOPICS = "topics";
 
   // Connector config

diff --git a/src/main/java/com/snowflake/kafka/connector/internal/streaming/SnowflakeSinkServiceV2.java b/src/main/java/com/snowflake/kafka/connector/internal/streaming/SnowflakeSinkServiceV2.java
@@ -203,8 +203,7 @@ private void createStreamingChannelForTopicPartition(
             this.sinkTaskContext,
             this.conn,
             this.recordService,
-            this.conn.getTelemetryClient(),
-            this.enableCustomJMXMonitoring));
+            this.conn.getTelemetryClient()));
   }
 
   /**

diff --git a/src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java b/src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java
@@ -8,19 +8,16 @@
 import static java.time.temporal.ChronoUnit.SECONDS;
 import static org.apache.kafka.common.record.TimestampType.NO_TIMESTAMP_TYPE;
 
-import com.codahale.metrics.MetricRegistry;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
-import com.snowflake.kafka.connector.SnowflakeSinkConnectorConfig;
 import com.snowflake.kafka.connector.Utils;
 import com.snowflake.kafka.connector.dlq.KafkaRecordErrorReporter;
 import com.snowflake.kafka.connector.internal.BufferThreshold;
 import com.snowflake.kafka.connector.internal.KCLogger;
 import com.snowflake.kafka.connector.internal.PartitionBuffer;
 import com.snowflake.kafka.connector.internal.SnowflakeConnectionService;
-import com.snowflake.kafka.connector.internal.metrics.MetricsJmxReporter;
 import com.snowflake.kafka.connector.internal.telemetry.SnowflakeTelemetryService;
 import com.snowflake.kafka.connector.records.RecordService;
 import com.snowflake.kafka.connector.records.SnowflakeJsonSchema;
@@ -175,15 +172,6 @@ public class TopicPartitionChannel {
   // Reference to the Snowflake connection service
   private final SnowflakeConnectionService conn;
 
-  // telemetry
-  private final SnowflakeTelemetryChannelStatus channelStatus;
-  // non null
-  private final MetricRegistry metricRegistry;
-
-  // Wrapper on Metric registry instance which will hold all registered metrics for this pipe
-  private final MetricsJmxReporter metricsJmxReporter;
-  private final boolean enableCustomJMXMonitoring;
-
   /**
    * Used to send telemetry to Snowflake. Currently, TelemetryClient created from a Snowflake
    * Connection Object, i.e. not a session-less Client
@@ -212,8 +200,7 @@ public TopicPartitionChannel(
         sinkTaskContext,
         null, /* Null Connection */
         new RecordService(null /* Null Telemetry Service*/),
-        null,
-        false);
+        null);
   }
 
   /**
@@ -230,8 +217,6 @@ public TopicPartitionChannel(
    * @param recordService record service for processing incoming offsets from Kafka
    * @param telemetryService Telemetry Service which includes the Telemetry Client, sends Json data
    *     to Snowflake
-   * @param enableCustomJMXMonitoring whether or not we enable Mbean for required classes and emit
-   *     JMX metrics for monitoring
    */
   public TopicPartitionChannel(
       SnowflakeStreamingIngestClient streamingIngestClient,
@@ -244,8 +229,7 @@ public TopicPartitionChannel(
       SinkTaskContext sinkTaskContext,
       SnowflakeConnectionService conn,
       RecordService recordService,
-      SnowflakeTelemetryService telemetryService,
-      boolean enableCustomJMXMonitoring) {
+      SnowflakeTelemetryService telemetryService) {
     this.streamingIngestClient = Preconditions.checkNotNull(streamingIngestClient);
     Preconditions.checkState(!streamingIngestClient.isClosed());
     this.topicPartition = Preconditions.checkNotNull(topicPartition);
@@ -292,25 +276,6 @@ public TopicPartitionChannel(
               + " correct offset instead",
           this.getChannelName());
     }
-
-    // jmx
-    this.enableCustomJMXMonitoring = enableCustomJMXMonitoring;
-    this.metricRegistry = new MetricRegistry();
-    this.metricsJmxReporter =
-        new MetricsJmxReporter(
-            this.metricRegistry,
-            this.sfConnectorConfig.getOrDefault(
-                SnowflakeSinkConnectorConfig.NAME, "noConnectorNameFound"));
-
-    // telemetry
-    this.channelStatus =
-        new SnowflakeTelemetryChannelStatus(
-            this.tableName,
-            this.topicPartition.topic(),
-            this.topicPartition.partition(),
-            this.channelName,
-            this.enableCustomJMXMonitoring,
-            this.metricsJmxReporter);
   }
 
   /**

diff --git a/...test/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannelTest.java b/...test/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannelTest.java
@@ -224,8 +224,7 @@ public void testCloseChannelException() throws Exception {
             mockSinkTaskContext,
             mockSnowflakeConnectionService,
             new RecordService(mockTelemetryService),
-            mockTelemetryService,
-            false);
+            mockTelemetryService);
 
     topicPartitionChannel.closeChannel();
   }
@@ -489,8 +488,7 @@ public void testInsertRowsWithSchemaEvolution() throws Exception {
               mockSinkTaskContext,
               conn,
               new RecordService(),
-              mockTelemetryService,
-              false);
+              mockTelemetryService);
 
       final int noOfRecords = 3;
       List<SinkRecord> records =
@@ -614,8 +612,7 @@ public void testInsertRows_ValidationResponseHasErrors_NoErrorTolerance() throws
             mockSinkTaskContext,
             mockSnowflakeConnectionService,
             new RecordService(mockTelemetryService),
-            mockTelemetryService,
-            false);
+            mockTelemetryService);
 
     List<SinkRecord> records = TestUtils.createJsonStringSinkRecords(0, 1, TOPIC, PARTITION);