S3 Dest emits V2 fields and captures failures (#42409)

Author: johnny-schmidt

airbyte-cdk/java/airbyte-cdk/README.md

@@ -26,7 +26,7 @@ constructor(
      * This probably doesn't belong here, but it's the easiest place where both [BufferEnqueue] and
      * [io.airbyte.cdk.integrations.destination.async.AsyncStreamConsumer] can both get to it.
      */
-    public val defaultNamespace: String?,
+    val defaultNamespace: String?,
     maxMemory: Long = (Runtime.getRuntime().maxMemory() * MEMORY_LIMIT_RATIO).toLong(),
 ) {
     @get:VisibleForTesting val buffers: ConcurrentMap<StreamDescriptor, StreamAwareQueue>

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/destination/async/buffers/BufferManager.kt

@@ -42,7 +42,11 @@ protected constructor(private val bufferStorage: BufferStorage) : SerializableBu
      */
     @Deprecated("")
     @Throws(IOException::class)
-    protected abstract fun writeRecord(record: AirbyteRecordMessage)
+    protected abstract fun writeRecord(
+        record: AirbyteRecordMessage,
+        generationId: Long = 0,
+        syncId: Long = 0
+    )
 
     /**
      * TODO: (ryankfu) move destination to use serialized record string instead of passing entire
@@ -80,7 +84,7 @@ protected constructor(private val bufferStorage: BufferStorage) : SerializableBu
 
     @Deprecated("")
     @Throws(Exception::class)
-    override fun accept(record: AirbyteRecordMessage): Long {
+    override fun accept(record: AirbyteRecordMessage, generationId: Long, syncId: Long): Long {
         if (!isStarted) {
             if (useCompression) {
                 compressedBuffer = GzipCompressorOutputStream(byteCounter)
@@ -92,7 +96,7 @@ protected constructor(private val bufferStorage: BufferStorage) : SerializableBu
         }
         if (inputStream == null && !isClosed) {
             val startCount = byteCounter.count
-            @Suppress("deprecation") writeRecord(record)
+            @Suppress("deprecation") writeRecord(record, generationId, syncId)
             return byteCounter.count - startCount
         } else {
             throw IllegalCallerException("Buffer is already closed, it cannot accept more messages")

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/destination/record_buffer/BaseSerializedBuffer.kt

@@ -29,7 +29,9 @@ interface BufferingStrategy : AutoCloseable {
     @Throws(Exception::class)
     fun addRecord(
         stream: AirbyteStreamNameNamespacePair,
-        message: AirbyteMessage
+        message: AirbyteMessage,
+        generationId: Long = 0,
+        syncId: Long = 0
     ): Optional<BufferFlushType>
 
     /** Flush the buffered messages from a single stream */

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/destination/record_buffer/BufferingStrategy.kt

@@ -41,7 +41,9 @@ class InMemoryRecordBufferingStrategy(
     @Throws(Exception::class)
     override fun addRecord(
         stream: AirbyteStreamNameNamespacePair,
-        message: AirbyteMessage
+        message: AirbyteMessage,
+        generationId: Long,
+        syncId: Long
     ): Optional<BufferFlushType> {
         var flushed: Optional<BufferFlushType> = Optional.empty()
 

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/destination/record_buffer/InMemoryRecordBufferingStrategy.kt

@@ -24,7 +24,9 @@ interface SerializableBuffer : AutoCloseable {
      * @param record [AirbyteRecordMessage] to be added to buffer
      * @return number of bytes written to the buffer
      */
-    @Deprecated("") @Throws(Exception::class) fun accept(record: AirbyteRecordMessage): Long
+    @Deprecated("")
+    @Throws(Exception::class)
+    fun accept(record: AirbyteRecordMessage, generationId: Long = 0, syncId: Long = 0): Long
 
     /**
      * TODO: (ryankfu) Move all destination connectors to pass the serialized record string instead

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/destination/record_buffer/SerializableBuffer.kt

@@ -47,12 +47,15 @@ class SerializedBufferingStrategy
     override fun addRecord(
         stream: AirbyteStreamNameNamespacePair,
         message: AirbyteMessage,
+        generationId: Long,
+        syncId: Long
     ): Optional<BufferFlushType> {
         var flushed: Optional<BufferFlushType> = Optional.empty()
 
         val buffer = getOrCreateBuffer(stream)
 
-        @Suppress("DEPRECATION") val actualMessageSizeInBytes = buffer.accept(message.record)
+        @Suppress("DEPRECATION")
+        val actualMessageSizeInBytes = buffer.accept(message.record, generationId, syncId)
         totalBufferSizeInBytes += actualMessageSizeInBytes
         // Flushes buffer when either the buffer was completely filled or only a single stream was
         // filled

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/destination/record_buffer/SerializedBufferingStrategy.kt

@@ -1 +1 @@
-version=0.44.9
+version=0.44.12

airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

@@ -24,6 +24,7 @@ import org.junit.jupiter.api.Test
 import org.mockito.ArgumentMatchers
 import org.mockito.Mockito
 import org.mockito.kotlin.any
+import org.mockito.kotlin.eq
 import org.mockito.kotlin.mock
 
 class BufferedStreamConsumerTest {
@@ -361,7 +362,7 @@ class BufferedStreamConsumerTest {
         // The first two records that we push will not trigger any flushes, but the third record
         // _will_
         // trigger a flush
-        Mockito.`when`(strategy.addRecord(any(), any()))
+        Mockito.`when`(strategy.addRecord(any(), any(), eq(0), eq(0)))
             .thenReturn(
                 Optional.empty(),
                 Optional.empty(),
@@ -463,7 +464,7 @@ class BufferedStreamConsumerTest {
         // The first two records that we push will not trigger any flushes, but the third record
         // _will_
         // trigger a flush
-        Mockito.`when`(strategy.addRecord(any(), any()))
+        Mockito.`when`(strategy.addRecord(any(), any(), eq(0), eq(0)))
             .thenReturn(
                 Optional.empty(),
                 Optional.empty(),

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/integrations/destination/buffered_stream_consumer/BufferedStreamConsumerTest.kt

@@ -14,6 +14,7 @@ import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.mockito.Mockito
 import org.mockito.kotlin.any
+import org.mockito.kotlin.eq
 
 class SerializedBufferingStrategyTest {
     private val catalog: ConfiguredAirbyteCatalog =
@@ -37,7 +38,7 @@ class SerializedBufferingStrategyTest {
 
     @Throws(Exception::class)
     private fun setupMock(mockObject: SerializableBuffer) {
-        Mockito.`when`(mockObject.accept(any())).thenReturn(10L)
+        Mockito.`when`(mockObject.accept(any(), eq(0L), eq(0L))).thenReturn(10L)
         Mockito.`when`(mockObject.byteCount).thenReturn(10L)
         Mockito.`when`(mockObject.maxTotalBufferSizeInBytes).thenReturn(MAX_TOTAL_BUFFER_SIZE_BYTES)
         Mockito.`when`(mockObject.maxPerStreamBufferSizeInBytes)