FLOW-7792 React to error deltas instead of error absolute value (#1240)

Author: sfc-gh-lkucharski

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

@@ -65,6 +65,11 @@ public class SnowpipeStreamingPartitionChannel implements TopicPartitionChannel
 
   private long lastAppendRowsOffset = NO_OFFSET_TOKEN_REGISTERED_IN_SNOWFLAKE;
 
+  // Tracks the initial error count when the channel was opened.
+  // Used to detect NEW errors (current error count > initial error count) since error counts
+  // are cumulative and don't reset when a channel is reopened.
+  private long initialErrorCount = 0;
+
   // Indicates whether we need to skip and discard any leftover rows in the current batch, this
   // could happen when the channel gets invalidated and reset
   private boolean needToSkipCurrentBatch = false;
@@ -490,23 +495,28 @@ private long fetchLatestOffsetFromChannel(SnowflakeStreamingIngestChannel channe
    *
    * @return new channel which was fetched after open/reopen
    */
-  private SnowflakeStreamingIngestChannel openChannelForTable(String channelName) {
-    SnowflakeStreamingIngestClient streamingIngestClient =
+  private SnowflakeStreamingIngestChannel openChannelForTable(final String channelName) {
+    final SnowflakeStreamingIngestClient streamingIngestClient =
         StreamingClientManager.getClient(
             connectorName, taskId, pipeName, connectorConfig, streamingClientProperties);
-    OpenChannelResult result = streamingIngestClient.openChannel(channelName, null);
-    if (result.getChannelStatus().getStatusCode().equals("SUCCESS")) {
-      LOGGER.info("Successfully opened streaming channel: {}", channelName);
+    final OpenChannelResult result = streamingIngestClient.openChannel(channelName, null);
+    final ChannelStatus channelStatus = result.getChannelStatus();
+    if (channelStatus.getStatusCode().equals("SUCCESS")) {
+      // Capture the initial error count - errors are cumulative and don't reset on channel reopen.
+      // We only want to fail on NEW errors that occur after the channel was opened.
+      this.initialErrorCount = channelStatus.getRowsErrorCount();
+      LOGGER.info(
+          "Successfully opened streaming channel: {}, initialErrorCount: {}",
+          channelName,
+          this.initialErrorCount);
       return result.getChannel();
     } else {
       LOGGER.error(
-          "Failed to open channel: {}, error code: {}",
-          channelName,
-          result.getChannelStatus().getStatusCode());
+          "Failed to open channel: {}, error code: {}", channelName, channelStatus.getStatusCode());
       throw ERROR_5028.getException(
           String.format(
               "Failed to open channel %s. Error code %s",
-              channelName, result.getChannelStatus().getStatusCode()));
+              channelName, channelStatus.getStatusCode()));
     }
   }
 
@@ -633,24 +643,36 @@ private void logChannelStatus(final ChannelStatus status) {
   }
 
   private void handleChannelErrors(final ChannelStatus status, final boolean tolerateErrors) {
-    final long rowsErrorCount = status.getRowsErrorCount();
-    if (rowsErrorCount > 0) {
+    final long currentErrorCount = status.getRowsErrorCount();
+    // Error counts are cumulative and don't reset when a channel is reopened.
+    // Only fail if there are NEW errors that occurred after the channel was opened.
+    final long newErrorCount = currentErrorCount - this.initialErrorCount;
+
+    if (newErrorCount > 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",
+              "Channel [%s] has %d new errors (total: %d, initial: %d). Last error message: %s,"
+                  + " last error timestamp: %s, last error offset token upper bound: %s",
               this.getChannelNameFormatV1(),
-              rowsErrorCount,
+              newErrorCount,
+              currentErrorCount,
+              this.initialErrorCount,
               status.getLastErrorMessage(),
               status.getLastErrorTimestamp(),
               status.getLastErrorOffsetTokenUpperBound());
 
+      this.initialErrorCount = currentErrorCount;
       if (tolerateErrors) {
         LOGGER.warn(errorMessage);
       } else {
         this.telemetryService.reportKafkaConnectFatalError(errorMessage);
         throw ERROR_5030.getException(errorMessage);
       }
+    } else if (currentErrorCount > 0) {
+      LOGGER.debug(
+          "Channel [{}] has {} pre-existing errors from before connector startup (no new errors)",
+          this.getChannelNameFormatV1(),
+          currentErrorCount);
     }
   }
 

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

@@ -174,6 +174,71 @@ void shouldContinueWorkingWhenChannelHasErrorsAndToleranceIsAll() throws JsonPro
         "All tasks should remain running when errors.tolerance=all");
   }
 
+  @Test
+  void shouldContinueWorkingWithPreExistingErrorsAndToleranceIsNone()
+      throws JsonProcessingException {
+    // Given: Pre-existing errors are set BEFORE the connector starts (simulating channel reopen
+    // scenario)
+    // This simulates the case where a channel has cumulative errors from a previous connector run
+    fakeClientSupplier.setPreExistingErrorCount(5);
+
+    Map<String, String> config = defaultProperties(topicName, connectorName);
+    config.put(KafkaConnectorConfigParams.ERRORS_TOLERANCE_CONFIG, "none");
+    connectCluster.configureConnector(connectorName, config);
+    waitForConnectorRunning(connectorName);
+
+    FakeSnowflakeStreamingIngestClient fakeClient = waitForConnectorToOpenChannels(connectorName);
+
+    // Produce messages
+    produceMessages(5);
+
+    // Then: connector should remain running because pre-existing errors don't count as new errors
+    await("Messages processed despite pre-existing errors")
+        .atMost(Duration.ofSeconds(30))
+        .until(() -> fakeClient.getAppendedRowCount() >= 5);
+
+    ConnectorStateInfo connectorState = connectCluster.connectorStatus(connectorName);
+    assertTrue(
+        connectorState.tasks().stream().allMatch(task -> "RUNNING".equals(task.state())),
+        "All tasks should be running when there are only pre-existing errors");
+  }
+
+  @Test
+  void shouldFailWhenNewErrorsOccurAfterStartupWithPreExistingErrors()
+      throws JsonProcessingException {
+    // Given: Pre-existing errors are set BEFORE the connector starts
+    fakeClientSupplier.setPreExistingErrorCount(5);
+
+    Map<String, String> config = defaultProperties(topicName, connectorName);
+    connectCluster.configureConnector(connectorName, config);
+    waitForConnectorRunning(connectorName);
+
+    FakeSnowflakeStreamingIngestClient fakeClient = waitForConnectorToOpenChannels(connectorName);
+
+    // Produce initial message
+    produceMessages(1);
+    await("Initial message processed")
+        .atMost(Duration.ofSeconds(30))
+        .until(() -> fakeClient.getAppendedRowCount() >= 1);
+
+    // When: NEW errors occur (error count increases from 5 to 10)
+    for (FakeSnowflakeStreamingIngestChannel channel : fakeClient.getOpenedChannels()) {
+      ChannelStatus statusWithNewErrors =
+          createChannelStatusWithErrors(channel.getChannelName(), 10);
+      channel.setChannelStatus(statusWithNewErrors);
+    }
+
+    // Then: connector task should fail due to NEW channel errors
+    await("Connector task failed due to new errors")
+        .atMost(Duration.ofMinutes(2))
+        .pollInterval(Duration.ofSeconds(4))
+        .until(
+            () -> {
+              ConnectorStateInfo state = connectCluster.connectorStatus(connectorName);
+              return state.tasks().stream().anyMatch(task -> "FAILED".equals(task.state()));
+            });
+  }
+
   private void produceMessages(int count) throws JsonProcessingException {
     Map<String, String> payload = Map.of("key1", "value1", "key2", "value2");
     for (int i = 0; i < count; i++) {

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

@@ -11,6 +11,8 @@ public class FakeIngestClientSupplier implements IngestClientSupplier {
   private final ConcurrentHashMap<String, FakeSnowflakeStreamingIngestClient>
       pipeToIngestClientMap = new ConcurrentHashMap<>();
 
+  private long preExistingErrorCount = 0;
+
   @Override
   public SnowflakeStreamingIngestClient get(
       final String clientName,
@@ -21,10 +23,27 @@ public SnowflakeStreamingIngestClient get(
       final StreamingClientProperties streamingClientProperties) {
     return pipeToIngestClientMap.computeIfAbsent(
         pipeName,
-        (key) -> new FakeSnowflakeStreamingIngestClient(pipeName, connectorConfig.get("name")));
+        (key) -> {
+          final FakeSnowflakeStreamingIngestClient client =
+              new FakeSnowflakeStreamingIngestClient(pipeName, connectorConfig.get("name"));
+          client.setDefaultErrorCount(preExistingErrorCount);
+          return client;
+        });
   }
 
   public Collection<FakeSnowflakeStreamingIngestClient> getFakeIngestClients() {
     return pipeToIngestClientMap.values();
   }
+
+  /**
+   * Sets the pre-existing error count that will be applied to all channels when they are opened.
+   * This simulates the cumulative error count that persists in Snowflake across connector restarts.
+   */
+  public void setPreExistingErrorCount(final long errorCount) {
+    this.preExistingErrorCount = errorCount;
+    // Also update existing clients
+    for (final FakeSnowflakeStreamingIngestClient client : pipeToIngestClientMap.values()) {
+      client.setDefaultErrorCount(errorCount);
+    }
+  }
 }

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

@@ -19,15 +19,21 @@ public class FakeSnowflakeStreamingIngestChannel
   private final String channelName;
   /** Collection of all rows appended to this channel */
   private final List<Map<String, Object>> appendedRows;
+  /** Reference to parent client for sharing error counts across channel reopens */
+  private final FakeSnowflakeStreamingIngestClient parentClient;
 
   private volatile boolean closed;
   private String offsetToken;
   private ChannelStatus channelStatus;
 
-  public FakeSnowflakeStreamingIngestChannel(String pipeName, String channelName) {
+  public FakeSnowflakeStreamingIngestChannel(
+      final String pipeName,
+      final String channelName,
+      final FakeSnowflakeStreamingIngestClient parentClient) {
     this.pipeName = pipeName;
     this.channelName = channelName;
     this.appendedRows = new ArrayList<>();
+    this.parentClient = parentClient;
   }
 
   @Override
@@ -109,6 +115,10 @@ public ChannelStatus getChannelStatus() {
 
   public void setChannelStatus(final ChannelStatus channelStatus) {
     this.channelStatus = channelStatus;
+    // Update the shared error count in the parent client so it persists across channel reopens
+    if (parentClient != null && channelStatus.getRowsErrorCount() > 0) {
+      parentClient.setInitialErrorCountForChannel(channelName, channelStatus.getRowsErrorCount());
+    }
   }
 
   @Override

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

@@ -20,13 +20,29 @@ public class FakeSnowflakeStreamingIngestClient implements SnowflakeStreamingIng
   private final List<FakeSnowflakeStreamingIngestChannel> openedChannels =
       Collections.synchronizedList(new ArrayList<>());
   private final Map<String, String> channelNameToOffsetTokens = new ConcurrentHashMap<>();
+  // Shared error counts per channel name - persists across channel reopens like real Snowflake
+  private final Map<String, Long> channelNameToErrorCount = new ConcurrentHashMap<>();
+  // Default error count to use when no channel-specific count is set
+  private long defaultErrorCount = 0;
   private boolean closed = false;
 
   public FakeSnowflakeStreamingIngestClient(final String pipeName, final String connectorName) {
     this.pipeName = pipeName;
     this.connectorName = connectorName;
   }
 
+  public void setDefaultErrorCount(final long errorCount) {
+    this.defaultErrorCount = errorCount;
+  }
+
+  public void setInitialErrorCountForChannel(final String channelName, final long errorCount) {
+    channelNameToErrorCount.put(channelName, errorCount);
+  }
+
+  public long getErrorCountForChannel(final String channelName) {
+    return channelNameToErrorCount.getOrDefault(channelName, defaultErrorCount);
+  }
+
   @Override
   public void close() {
     this.closed = true;
@@ -52,6 +68,9 @@ public OpenChannelResult openChannel(final String channelName, final String offs
     if (offsetToken != null) {
       channelNameToOffsetTokens.put(channelName, offsetToken);
     }
+    // Use the shared error count - this persists across channel reopens like real Snowflake
+    // Falls back to defaultErrorCount if no channel-specific count is set
+    final long errorCount = channelNameToErrorCount.getOrDefault(channelName, defaultErrorCount);
     final ChannelStatus channelStatus =
         new ChannelStatus(
             "db",
@@ -63,14 +82,14 @@ public OpenChannelResult openChannel(final String channelName, final String offs
             Instant.now(),
             0, // rowsInsertedCount
             0, // rowsParsedCount
-            0, // rowsErrorCount - default to 0 (no errors)
+            errorCount, // rowsErrorCount - use shared error count
             null,
             null,
             null,
             null,
             Instant.now());
     final FakeSnowflakeStreamingIngestChannel channel =
-        new FakeSnowflakeStreamingIngestChannel(pipeName, channelName);
+        new FakeSnowflakeStreamingIngestChannel(pipeName, channelName, this);
     channel.setChannelStatus(channelStatus); // Set default channel status
     openedChannels.add(channel);
     return new OpenChannelResult(channel, channelStatus);

src/test/java/com/snowflake/kafka/connector/streaming/iceberg/IcebergIngestionIntoVariantIT.java

@@ -12,8 +12,10 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
+@Disabled("Regresion introduced on production FLOW-7864")
 public class IcebergIngestionIntoVariantIT extends IcebergIngestionIT {
 
   @Test

src/test/java/com/snowflake/kafka/connector/streaming/iceberg/IcebergInitServiceIT.java

@@ -8,6 +8,7 @@
 import com.snowflake.kafka.connector.internal.TestUtils;
 import org.junit.jupiter.api.*;
 
+@Disabled("Regresion introduced on production FLOW-7864")
 public class IcebergInitServiceIT extends BaseIcebergIT {
 
   private static IcebergInitService icebergInitService;