Rcheng doublebuff2

src/main/java/com/snowflake/kafka/connector/internal/metrics/MetricsUtil.java

@@ -89,8 +89,6 @@ public class MetricsUtil {
    * description
    */
   public static final String OFFSET_PERSISTED_IN_SNOWFLAKE = "persisted-in-snowflake-offset";
-
-  public static final String LATEST_CONSUMER_OFFSET = "latest-consumer-offset";
   // ********** ^ Streaming Constants ^ **********//
 
   public enum EventType {

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

@@ -267,7 +267,6 @@ private void createStreamingChannelForTopicPartition(
             partitionChannelKey, // Streaming channel name
             tableName,
             hasSchemaEvolutionPermission,
-            new StreamingBufferThreshold(this.flushTimeSeconds, this.fileSizeBytes, this.recordNum),
             this.connectorConfig,
             this.kafkaRecordErrorReporter,
             this.sinkTaskContext,
@@ -300,12 +299,6 @@ public void insert(Collection<SinkRecord> records) {
       // threshold.
       insert(record);
     }
-
-    // check all partitions to see if they need to be flushed based on time
-    for (TopicPartitionChannel partitionChannel : partitionsToChannel.values()) {
-      // Time based flushing
-      partitionChannel.insertBufferedRecordsIfFlushTimeThresholdReached();
-    }
   }
 
   /**
@@ -335,7 +328,7 @@ public void insert(SinkRecord record) {
     }
 
     TopicPartitionChannel channelPartition = partitionsToChannel.get(partitionChannelKey);
-    channelPartition.insertRecordToBuffer(record);
+    channelPartition.insertRecord(record);
   }
 
   @Override
@@ -348,7 +341,6 @@ public long getOffset(TopicPartition topicPartition) {
             this.shouldUseConnectorNameInChannelName);
     if (partitionsToChannel.containsKey(partitionChannelKey)) {
       long offset = partitionsToChannel.get(partitionChannelKey).getOffsetSafeToCommitToKafka();
-      partitionsToChannel.get(partitionChannelKey).setLatestConsumerOffset(offset);
       return offset;
     } else {
       LOGGER.warn(