FLOW-7395 Use channel status error count

Author: sfc-gh-lkucharski

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

@@ -25,6 +25,7 @@
 import com.snowflake.kafka.connector.internal.SnowflakeConnectionService;
 import com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceFactory;
 import com.snowflake.kafka.connector.internal.SnowflakeErrors;
+import com.snowflake.kafka.connector.internal.SnowflakeKafkaConnectorException;
 import com.snowflake.kafka.connector.internal.SnowflakeSinkService;
 import com.snowflake.kafka.connector.internal.streaming.SnowflakeSinkServiceV2;
 import java.util.Arrays;
@@ -77,6 +78,9 @@ public class SnowflakeSinkTask extends SinkTask {
   private final SnowflakeSinkTaskAuthorizationExceptionTracker authorizationExceptionTracker =
       new SnowflakeSinkTaskAuthorizationExceptionTracker();
 
+  // Stores channel error exception detected in preCommit to fail on next put() call
+  private volatile SnowflakeKafkaConnectorException channelErrorToFailOn = null;
+
   /** default constructor, invoked by kafka connect framework */
   public SnowflakeSinkTask() {
     DYNAMIC_LOGGER = new KCLogger(this.getClass().getName());
@@ -266,6 +270,13 @@ public void close(final Collection<TopicPartition> partitions) {
   public void put(final Collection<SinkRecord> records) {
     this.authorizationExceptionTracker.throwExceptionIfAuthorizationFailed();
 
+    // Check for channel errors detected in preCommit and fail the task
+    if (this.channelErrorToFailOn != null) {
+      SnowflakeKafkaConnectorException error = this.channelErrorToFailOn;
+      this.channelErrorToFailOn = null; // Clear so we don't throw again on restart
+      throw new ConnectException(error.getMessage(), error);
+    }
+
     final long recordSize = records.size();
     DYNAMIC_LOGGER.debug("Calling PUT with {} records", recordSize);
 
@@ -319,6 +330,13 @@ public Map<TopicPartition, OffsetAndMetadata> preCommit(
               committedOffsets.put(topicPartition, new OffsetAndMetadata(offset));
             }
           });
+    } catch (SnowflakeKafkaConnectorException e) {
+      this.authorizationExceptionTracker.reportPrecommitException(e);
+      this.DYNAMIC_LOGGER.error("PreCommit error: {} ", e.getMessage());
+      // Channel error count exceeded - store to fail on next put() call
+      if (e.checkErrorCode(SnowflakeErrors.ERROR_5030)) {
+        this.channelErrorToFailOn = e;
+      }
     } catch (Exception e) {
       this.authorizationExceptionTracker.reportPrecommitException(e);
       this.DYNAMIC_LOGGER.error("PreCommit error: {} ", e.getMessage());

src/main/java/com/snowflake/kafka/connector/internal/SnowflakeErrors.java

@@ -233,7 +233,11 @@ public enum SnowflakeErrors {
       "5029",
       "Destination table does not exist",
       "Destination table does not exist. Please ensure the destination table exists in Snowflake"
-          + " before starting the connector.");
+          + " before starting the connector."),
+  ERROR_5030(
+      "5030",
+      "Channel error count threshold exceeded",
+      "Channel has reported errors during data ingestion. Check the channel history for details.");
 
   // properties
 

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

@@ -81,6 +81,8 @@ public class SnowflakeSinkServiceV2 implements SnowflakeSinkService {
   // default is true unless the configuration provided is false;
   // If this is true, we will enable Mbean for required classes and emit JMX metrics for monitoring
   private boolean enableCustomJMXMonitoring = KafkaConnectorConfigParams.JMX_OPT_DEFAULT;
+  // Whether to tolerate errors during ingestion (based on errors.tolerance config)
+  private final boolean tolerateErrors;
 
   public SnowflakeSinkServiceV2(
       SnowflakeConnectionService conn,
@@ -118,11 +120,13 @@ public SnowflakeSinkServiceV2(
     }
 
     this.metricsJmxReporter = new MetricsJmxReporter(new MetricRegistry(), this.connectorName);
+    this.tolerateErrors = StreamingUtils.tolerateErrors(connectorConfig);
 
     LOGGER.info(
-        "SnowflakeSinkServiceV2 initialized for connector: {}, task: {}",
+        "SnowflakeSinkServiceV2 initialized for connector: {}, task: {}, tolerateErrors: {}",
         this.connectorName,
-        this.taskId);
+        this.taskId,
+        this.tolerateErrors);
   }
 
   /** Gets a unique identifier consisting of connector name, topic name and partition number. */
@@ -322,8 +326,10 @@ public long getOffset(TopicPartition topicPartition) {
     String partitionChannelKey =
         makeChannelName(this.connectorName, topicPartition.topic(), topicPartition.partition());
     if (partitionsToChannel.containsKey(partitionChannelKey)) {
-      long offset = partitionsToChannel.get(partitionChannelKey).getOffsetSafeToCommitToKafka();
-      partitionsToChannel.get(partitionChannelKey).setLatestConsumerGroupOffset(offset);
+      TopicPartitionChannel channel = partitionsToChannel.get(partitionChannelKey);
+      channel.checkChannelStatusAndLogErrors(tolerateErrors);
+      long offset = channel.getOffsetSafeToCommitToKafka();
+      channel.setLatestConsumerGroupOffset(offset);
       LOGGER.info(
           "Fetched snowflake commited offset: [{}] for channel [{}]", offset, partitionChannelKey);
       return offset;

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

@@ -68,6 +68,8 @@ default long getOffsetSafeToCommitToKafka() {
 
   void setLatestConsumerGroupOffset(long consumerOffset);
 
+  void checkChannelStatusAndLogErrors(boolean tolerateErrors);
+
   default CompletableFuture<Void> waitForLastProcessedRecordCommitted() {
     return CompletableFuture.completedFuture(null);
   }

src/main/java/com/snowflake/kafka/connector/internal/streaming/v2/SnowpipeStreamingPartitionChannel.java

@@ -2,8 +2,10 @@
 
 import static com.snowflake.kafka.connector.internal.SnowflakeErrors.ERROR_5027;
 import static com.snowflake.kafka.connector.internal.SnowflakeErrors.ERROR_5028;
+import static com.snowflake.kafka.connector.internal.SnowflakeErrors.ERROR_5030;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.snowflake.ingest.streaming.ChannelStatus;
 import com.snowflake.ingest.streaming.OpenChannelResult;
 import com.snowflake.ingest.streaming.SFException;
 import com.snowflake.ingest.streaming.SnowflakeStreamingIngestChannel;
@@ -583,6 +585,61 @@ public void setLatestConsumerGroupOffset(long consumerOffset) {
     }
   }
 
+  @Override
+  public void checkChannelStatusAndLogErrors(final boolean tolerateErrors) {
+    final ChannelStatus status = channel.getChannelStatus();
+    logChannelStatus(status);
+    handleChannelErrors(status, tolerateErrors);
+  }
+
+  private void logChannelStatus(final ChannelStatus status) {
+    LOGGER.info(
+        "Channel status for channel=[{}]: databaseName=[{}], schemaName=[{}], pipeName=[{}],"
+            + " channelName=[{}], statusCode=[{}], latestCommittedOffsetToken=[{}],"
+            + " createdOn=[{}], rowsInsertedCount=[{}], rowsParsedCount=[{}],"
+            + " rowsErrorCount=[{}], lastErrorOffsetTokenUpperBound=[{}],"
+            + " lastErrorMessage=[{}], lastErrorTimestamp=[{}],"
+            + " serverAvgProcessingLatency=[{}], lastRefreshedOn=[{}]",
+        this.getChannelNameFormatV1(),
+        status.getDatabaseName(),
+        status.getSchemaName(),
+        status.getPipeName(),
+        status.getChannelName(),
+        status.getStatusCode(),
+        status.getLatestCommittedOffsetToken(),
+        status.getCreatedOn(),
+        status.getRowsInsertedCount(),
+        status.getRowsParsedCount(),
+        status.getRowsErrorCount(),
+        status.getLastErrorOffsetTokenUpperBound(),
+        status.getLastErrorMessage(),
+        status.getLastErrorTimestamp(),
+        status.getServerAvgProcessingLatency(),
+        status.getLastRefreshedOn());
+  }
+
+  private void handleChannelErrors(final ChannelStatus status, final boolean tolerateErrors) {
+    final long rowsErrorCount = status.getRowsErrorCount();
+    if (rowsErrorCount > 0) {
+      final String errorMessage =
+          String.format(
+              "Channel [%s] has %d errors. Last error message: %s, last error timestamp: %s,"
+                  + " last error offset token upper bound: %s",
+              this.getChannelNameFormatV1(),
+              rowsErrorCount,
+              status.getLastErrorMessage(),
+              status.getLastErrorTimestamp(),
+              status.getLastErrorOffsetTokenUpperBound());
+
+      if (tolerateErrors) {
+        LOGGER.warn(errorMessage);
+      } else {
+        this.telemetryServiceV2.reportKafkaConnectFatalError(errorMessage);
+        throw ERROR_5030.getException(errorMessage);
+      }
+    }
+  }
+
   /**
    * Enum representing which Streaming API is invoking the fallback supplier. ({@link
    * #streamingApiFallbackSupplier(StreamingApiFallbackInvoker)})