SNOW-1514185 Assign new channel when no offset is present in Snowflake

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

@@ -905,6 +905,7 @@ private void resetChannelMetadataAfterRecovery(
  ? latestConsumerOffset.get()
  : offsetRecoveredFromSnowflake + 1L;
  if (offsetToResetInKafka == NO_OFFSET_TOKEN_REGISTERED_IN_SNOWFLAKE) {
+ this.channel = newChannel;
  return;
  }
 

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

@@ -707,6 +707,7 @@ private void resetChannelMetadataAfterRecovery(
  ? currentConsumerGroupOffset.get()
  : offsetRecoveredFromSnowflake + 1L;
  if (offsetToResetInKafka == NO_OFFSET_TOKEN_REGISTERED_IN_SNOWFLAKE) {
+ this.channel = newChannel;
  return;
  }
 

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

@@ -1311,4 +1311,46 @@ private void insertAndFlush(TopicPartitionChannel channel, List<SinkRecord> reco
  Thread.sleep(this.streamingBufferThreshold.getFlushTimeThresholdSeconds() + 1);
  channel.insertBufferedRecordsIfFlushTimeThresholdReached();
  }
+
+ @Test
+ public void assignANewChannel_whenNoOffsetIsPresentInSnowflake() {
+ // given
+ String noOffset = "-1";
+
+ SnowflakeStreamingIngestChannel originalChannel =
+ Mockito.mock(SnowflakeStreamingIngestChannel.class);
+ Mockito.when(originalChannel.getLatestCommittedOffsetToken())
+ .thenReturn(noOffset)
+ .thenThrow(new SFException(ErrorCode.CHANNEL_STATUS_INVALID));
+
+ SnowflakeStreamingIngestChannel reopenedChannel =
+ Mockito.mock(SnowflakeStreamingIngestChannel.class);
+ Mockito.when(reopenedChannel.getLatestCommittedOffsetToken()).thenReturn(noOffset);
+
+ Mockito.when(mockStreamingClient.openChannel(any(OpenChannelRequest.class)))
+ .thenReturn(originalChannel, reopenedChannel);
+
+ TopicPartitionChannel topicPartitionChannel =
+ createTopicPartitionChannel(
+ this.mockStreamingClient,
+ this.topicPartition,
+ TEST_CHANNEL_NAME,
+ TEST_TABLE_NAME,
+ this.enableSchematization,
+ this.streamingBufferThreshold,
+ this.sfConnectorConfig,
+ this.mockKafkaRecordErrorReporter,
+ this.mockSinkTaskContext,
+ this.mockSnowflakeConnectionService,
+ new RecordService(),
+ this.mockTelemetryService,
+ true,
+ null);
+
+ // when
+ topicPartitionChannel.getOffsetSafeToCommitToKafka();
+
+ // then
+ Assert.assertEquals(reopenedChannel, topicPartitionChannel.getChannel());
+ }
 }