fix(cdc): limit queue size to lower memory consumption

airbyte-integrations/bases/debezium/src/main/java/io/airbyte/integrations/debezium/AirbyteDebeziumHandler.java

@@ -18,6 +18,7 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Optional;
+import java.util.OptionalInt;
 import java.util.Properties;
 import java.util.concurrent.LinkedBlockingQueue;
 import org.slf4j.Logger;
@@ -41,15 +42,18 @@ public class AirbyteDebeziumHandler<T> {
   private final CdcTargetPosition<T> targetPosition;
   private final boolean trackSchemaHistory;
   private final Duration firstRecordWaitTime;
+  private final OptionalInt queueSize;
 
   public AirbyteDebeziumHandler(final JsonNode config,
                                 final CdcTargetPosition<T> targetPosition,
                                 final boolean trackSchemaHistory,
-                                final Duration firstRecordWaitTime) {
+                                final Duration firstRecordWaitTime,
+                                final OptionalInt queueSize) {
     this.config = config;
     this.targetPosition = targetPosition;
     this.trackSchemaHistory = trackSchemaHistory;
     this.firstRecordWaitTime = firstRecordWaitTime;
+    this.queueSize = queueSize;
   }
 
   public AutoCloseableIterator<AirbyteMessage> getSnapshotIterators(
@@ -60,7 +64,7 @@ public AutoCloseableIterator<AirbyteMessage> getSnapshotIterators(
                                                                     final Instant emittedAt) {
 
     LOGGER.info("Running snapshot for " + catalogContainingStreamsToSnapshot.getStreams().size() + " new tables");
-    final LinkedBlockingQueue<ChangeEvent<String, String>> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
+    final LinkedBlockingQueue<ChangeEvent<String, String>> queue = new LinkedBlockingQueue<>(queueSize.orElse(QUEUE_CAPACITY));
 
     final AirbyteFileOffsetBackingStore offsetManager = AirbyteFileOffsetBackingStore.initializeDummyStateForSnapshotPurpose();
     final DebeziumRecordPublisher tableSnapshotPublisher = new DebeziumRecordPublisher(snapshotProperties,
@@ -93,7 +97,7 @@ public AutoCloseableIterator<AirbyteMessage> getIncrementalIterators(final Confi
                                                                        final Instant emittedAt,
                                                                        final boolean addDbNameToState) {
     LOGGER.info("Using CDC: {}", true);
-    final LinkedBlockingQueue<ChangeEvent<String, String>> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
+    final LinkedBlockingQueue<ChangeEvent<String, String>> queue = new LinkedBlockingQueue<>(queueSize.orElse(QUEUE_CAPACITY));
     final AirbyteFileOffsetBackingStore offsetManager = AirbyteFileOffsetBackingStore.initializeState(cdcSavedInfoFetcher.getSavedOffset(),
         addDbNameToState ? Optional.ofNullable(config.get(JdbcUtils.DATABASE_KEY).asText()) : Optional.empty());
     final Optional<AirbyteSchemaHistoryStorage> schemaHistoryManager = schemaHistoryManager(cdcSavedInfoFetcher);

airbyte-integrations/bases/debezium/src/main/java/io/airbyte/integrations/debezium/internals/DebeziumPropertiesManager.java

@@ -22,6 +22,7 @@
 public class DebeziumPropertiesManager {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DebeziumPropertiesManager.class);
+  private static final String BYTE_VALUE_256_MB = Integer.toString(256 * 1024 * 1024);
   private final JsonNode config;
   private final AirbyteFileOffsetBackingStore offsetManager;
   private final Optional<AirbyteSchemaHistoryStorage> schemaHistoryManager;
@@ -46,30 +47,33 @@ public Properties getDebeziumProperties() {
     props.putAll(properties);
 
     // debezium engine configuration
-    // https://debezium.io/documentation/reference/2.1/development/engine.html#engine-properties
+    // https://debezium.io/documentation/reference/2.2/development/engine.html#engine-properties
     props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
     props.setProperty("offset.storage.file.filename", offsetManager.getOffsetFilePath().toString());
     props.setProperty("offset.flush.interval.ms", "1000"); // todo: make this longer
     // default values from debezium CommonConnectorConfig
     props.setProperty("max.batch.size", "2048");
     props.setProperty("max.queue.size", "8192");
 
-    props.setProperty("errors.max.retries", "10");
+    // Disabling retries because debezium startup time might exceed our 60-second wait limit
+    // The maximum number of retries on connection errors before failing (-1 = no limit, 0 = disabled, >
+    // 0 = num of retries).
+    props.setProperty("errors.max.retries", "0");
     // This property must be strictly less than errors.retry.delay.max.ms
     // (https://github.com/debezium/debezium/blob/bcc7d49519a4f07d123c616cfa45cd6268def0b9/debezium-core/src/main/java/io/debezium/util/DelayStrategy.java#L135)
     props.setProperty("errors.retry.delay.initial.ms", "299");
     props.setProperty("errors.retry.delay.max.ms", "300");
 
     if (schemaHistoryManager.isPresent()) {
-      // https://debezium.io/documentation/reference/2.1/operations/debezium-server.html#debezium-source-database-history-class
+      // https://debezium.io/documentation/reference/2.2/operations/debezium-server.html#debezium-source-database-history-class
       // https://debezium.io/documentation/reference/development/engine.html#_in_the_code
       // As mentioned in the documents above, debezium connector for MySQL needs to track the schema
       // changes. If we don't do this, we can't fetch records for the table.
       props.setProperty("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory");
       props.setProperty("schema.history.internal.file.filename", schemaHistoryManager.get().getPath().toString());
     }
 
-    // https://debezium.io/documentation/reference/2.1/configuration/avro.html
+    // https://debezium.io/documentation/reference/2.2/configuration/avro.html
     props.setProperty("key.converter.schemas.enable", "false");
     props.setProperty("value.converter.schemas.enable", "false");
 
@@ -89,10 +93,13 @@ public Properties getDebeziumProperties() {
     // By default "decimal.handing.mode=precise" which's caused returning this value as a binary.
     // The "double" type may cause a loss of precision, so set Debezium's config to store it as a String
     // explicitly in its Kafka messages for more details see:
-    // https://debezium.io/documentation/reference/2.1/connectors/postgresql.html#postgresql-decimal-types
+    // https://debezium.io/documentation/reference/2.2/connectors/postgresql.html#postgresql-decimal-types
     // https://debezium.io/documentation/faq/#how_to_retrieve_decimal_field_from_binary_representation
     props.setProperty("decimal.handling.mode", "string");
 
+    // https://debezium.io/documentation/reference/2.2/connectors/postgresql.html#postgresql-property-max-queue-size-in-bytes
+    props.setProperty("max.queue.size.in.bytes", BYTE_VALUE_256_MB);
+
     // WARNING : Never change the value of this otherwise all the connectors would start syncing from
     // scratch
     props.setProperty("topic.prefix", config.get(JdbcUtils.DATABASE_KEY).asText());

airbyte-integrations/bases/debezium/src/main/java/io/airbyte/integrations/debezium/internals/DebeziumRecordPublisher.java

@@ -59,7 +59,7 @@ public void start(final BlockingQueue<ChangeEvent<String, String>> queue) {
           // debezium outputs a tombstone event that has a value of null. this is an artifact of how it
           // interacts with kafka. we want to ignore it.
           // more on the tombstone:
-          // https://debezium.io/documentation/reference/2.1/transformations/event-flattening.html
+          // https://debezium.io/documentation/reference/2.2/transformations/event-flattening.html
           if (e.value() != null) {
             try {
               queue.put(e);

airbyte-integrations/bases/debezium/src/main/java/io/airbyte/integrations/debezium/internals/postgres/PostgresConverter.java

@@ -240,7 +240,7 @@ private int getTimePrecision(final RelationalColumn field) {
   }
 
   // Ref :
-  // https://debezium.io/documentation/reference/2.1/connectors/postgresql.html#postgresql-temporal-types
+  // https://debezium.io/documentation/reference/2.2/connectors/postgresql.html#postgresql-temporal-types
   private void registerDate(final RelationalColumn field, final ConverterRegistration<SchemaBuilder> registration) {
     final var fieldType = field.typeName();
 

airbyte-integrations/connectors/source-mssql-strict-encrypt/Dockerfile

@@ -24,5 +24,5 @@ ENV APPLICATION source-mssql-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=1.0.16
+LABEL io.airbyte.version=1.0.17
 LABEL io.airbyte.name=airbyte/source-mssql-strict-encrypt

airbyte-integrations/connectors/source-mssql-strict-encrypt/metadata.yaml

@@ -11,7 +11,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
-  dockerImageTag: 1.0.16
+  dockerImageTag: 1.0.17
   dockerRepository: airbyte/source-mssql-strict-encrypt
   githubIssueLabel: source-mssql
   icon: mssql.svg

airbyte-integrations/connectors/source-mssql/Dockerfile

@@ -24,5 +24,5 @@ ENV APPLICATION source-mssql
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=1.0.16
+LABEL io.airbyte.version=1.0.17
 LABEL io.airbyte.name=airbyte/source-mssql

airbyte-integrations/connectors/source-mssql/metadata.yaml

@@ -6,7 +6,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
-  dockerImageTag: 1.0.16
+  dockerImageTag: 1.0.17
   dockerRepository: airbyte/source-mssql
   githubIssueLabel: source-mssql
   icon: mssql.svg

airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcHelper.java

@@ -39,7 +39,7 @@ public enum ReplicationMethod {
    * The default "SNAPSHOT" mode can prevent other (non-Airbyte) transactions from updating table rows
    * while we snapshot. References:
    * https://docs.microsoft.com/en-us/sql/t-sql/statements/set-transaction-isolation-level-transact-sql?view=sql-server-ver15
-   * https://debezium.io/documentation/reference/2.1/connectors/sqlserver.html#sqlserver-property-snapshot-isolation-mode
+   * https://debezium.io/documentation/reference/2.2/connectors/sqlserver.html#sqlserver-property-snapshot-isolation-mode
    */
   public enum SnapshotIsolation {
 
@@ -69,7 +69,7 @@ public static SnapshotIsolation from(final String jsonValue) {
 
   }
 
-  // https://debezium.io/documentation/reference/2.1/connectors/sqlserver.html#sqlserver-property-snapshot-mode
+  // https://debezium.io/documentation/reference/2.2/connectors/sqlserver.html#sqlserver-property-snapshot-mode
   public enum DataToSync {
 
     EXISTING_AND_NEW("Existing and New", "initial"),
@@ -146,9 +146,9 @@ static Properties getDebeziumProperties(final JdbcDatabase database, final Confi
     final Properties props = new Properties();
     props.setProperty("connector.class", "io.debezium.connector.sqlserver.SqlServerConnector");
 
-    // https://debezium.io/documentation/reference/2.1/connectors/sqlserver.html#sqlserver-property-include-schema-changes
+    // https://debezium.io/documentation/reference/2.2/connectors/sqlserver.html#sqlserver-property-include-schema-changes
     props.setProperty("include.schema.changes", "false");
-    // https://debezium.io/documentation/reference/2.1/connectors/sqlserver.html#sqlserver-property-provide-transaction-metadata
+    // https://debezium.io/documentation/reference/2.2/connectors/sqlserver.html#sqlserver-property-provide-transaction-metadata
     props.setProperty("provide.transaction.metadata", "false");
 
     props.setProperty("converters", "mssql_converter");

airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSource.java

@@ -59,6 +59,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.OptionalInt;
 import java.util.Set;
 import java.util.function.Supplier;
 import org.slf4j.Logger;
@@ -449,7 +450,7 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(
       final Duration firstRecordWaitTime = FirstRecordWaitTimeUtil.getFirstRecordWaitTime(sourceConfig);
       final AirbyteDebeziumHandler<Lsn> handler =
           new AirbyteDebeziumHandler<>(sourceConfig,
-              MssqlCdcTargetPosition.getTargetPosition(database, sourceConfig.get(JdbcUtils.DATABASE_KEY).asText()), true, firstRecordWaitTime);
+              MssqlCdcTargetPosition.getTargetPosition(database, sourceConfig.get(JdbcUtils.DATABASE_KEY).asText()), true, firstRecordWaitTime, OptionalInt.empty());
 
       final Supplier<AutoCloseableIterator<AirbyteMessage>> incrementalIteratorSupplier = () -> handler.getIncrementalIterators(catalog,
           new MssqlCdcSavedInfoFetcher(stateManager.getCdcStateManager().getCdcState()),

airbyte-integrations/connectors/source-mysql-strict-encrypt/Dockerfile

@@ -24,6 +24,6 @@ ENV APPLICATION source-mysql-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=2.0.23
+LABEL io.airbyte.version=2.0.24
 
 LABEL io.airbyte.name=airbyte/source-mysql-strict-encrypt

airbyte-integrations/connectors/source-mysql-strict-encrypt/metadata.yaml

@@ -11,7 +11,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
-  dockerImageTag: 2.0.23
+  dockerImageTag: 2.0.24
   dockerRepository: airbyte/source-mysql-strict-encrypt
   githubIssueLabel: source-mysql
   icon: mysql.svg

airbyte-integrations/connectors/source-mysql/Dockerfile

@@ -24,6 +24,6 @@ ENV APPLICATION source-mysql
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=2.0.23
+LABEL io.airbyte.version=2.0.24
 
 LABEL io.airbyte.name=airbyte/source-mysql

airbyte-integrations/connectors/source-mysql/metadata.yaml

@@ -6,7 +6,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
-  dockerImageTag: 2.0.23
+  dockerImageTag: 2.0.24
   dockerRepository: airbyte/source-mysql
   githubIssueLabel: source-mysql
   icon: mysql.svg

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlCdcProperties.java

@@ -37,7 +37,7 @@ static Properties getDebeziumProperties(final JdbcDatabase database) {
       // initial snapshot
       props.setProperty("snapshot.mode", sourceConfig.get("snapshot_mode").asText());
     } else {
-      // https://debezium.io/documentation/reference/2.1/connectors/mysql.html#mysql-property-snapshot-mode
+      // https://debezium.io/documentation/reference/2.2/connectors/mysql.html#mysql-property-snapshot-mode
       props.setProperty("snapshot.mode", "when_needed");
     }
 
@@ -52,8 +52,8 @@ private static Properties commonProperties(final JdbcDatabase database) {
     props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
 
     props.setProperty("database.server.id", String.valueOf(generateServerID()));
-    // https://debezium.io/documentation/reference/2.1/connectors/mysql.html#mysql-boolean-values
-    // https://debezium.io/documentation/reference/2.1/development/converters.html
+    // https://debezium.io/documentation/reference/2.2/connectors/mysql.html#mysql-boolean-values
+    // https://debezium.io/documentation/reference/2.2/development/converters.html
     /**
      * {@link io.debezium.connector.mysql.converters.TinyIntOneToBooleanConverter}
      * {@link MySQLConverter}
@@ -74,7 +74,7 @@ private static Properties commonProperties(final JdbcDatabase database) {
     }
 
     // Check params for SSL connection in config and add properties for CDC SSL connection
-    // https://debezium.io/documentation/reference/2.1/connectors/mysql.html#mysql-property-database-ssl-mode
+    // https://debezium.io/documentation/reference/2.2/connectors/mysql.html#mysql-property-database-ssl-mode
     if (!sourceConfig.has(JdbcUtils.SSL_KEY) || sourceConfig.get(JdbcUtils.SSL_KEY).asBoolean()) {
       if (dbConfig.has(SSL_MODE) && !dbConfig.get(SSL_MODE).asText().isEmpty()) {
         props.setProperty("database.ssl.mode", MySqlSource.toSslJdbcParamInternal(SslMode.valueOf(dbConfig.get(SSL_MODE).asText())));
@@ -100,15 +100,15 @@ private static Properties commonProperties(final JdbcDatabase database) {
       }
     }
 
-    // https://debezium.io/documentation/reference/2.1/connectors/mysql.html#mysql-property-snapshot-locking-mode
+    // https://debezium.io/documentation/reference/2.2/connectors/mysql.html#mysql-property-snapshot-locking-mode
     // This is to make sure other database clients are allowed to write to a table while Airbyte is
     // taking a snapshot. There is a risk involved that
     // if any database client makes a schema change then the sync might break
     props.setProperty("snapshot.locking.mode", "none");
-    // https://debezium.io/documentation/reference/2.1/connectors/mysql.html#mysql-property-include-schema-changes
+    // https://debezium.io/documentation/reference/2.2/connectors/mysql.html#mysql-property-include-schema-changes
     props.setProperty("include.schema.changes", "false");
     // This to make sure that binary data represented as a base64-encoded String.
-    // https://debezium.io/documentation/reference/2.1/connectors/mysql.html#mysql-property-binary-handling-mode
+    // https://debezium.io/documentation/reference/2.2/connectors/mysql.html#mysql-property-binary-handling-mode
     props.setProperty("binary.handling.mode", "base64");
     props.setProperty("database.include.list", sourceConfig.get("database").asText());
 

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlSource.java

@@ -67,6 +67,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.OptionalInt;
 import java.util.Set;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -274,7 +275,7 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(final
       final Duration firstRecordWaitTime = FirstRecordWaitTimeUtil.getFirstRecordWaitTime(sourceConfig);
       LOGGER.info("First record waiting time: {} seconds", firstRecordWaitTime.getSeconds());
       final AirbyteDebeziumHandler<MySqlCdcPosition> handler =
-          new AirbyteDebeziumHandler<>(sourceConfig, MySqlCdcTargetPosition.targetPosition(database), true, firstRecordWaitTime);
+          new AirbyteDebeziumHandler<>(sourceConfig, MySqlCdcTargetPosition.targetPosition(database), true, firstRecordWaitTime, OptionalInt.empty());
 
       final MySqlCdcStateHandler mySqlCdcStateHandler = new MySqlCdcStateHandler(stateManager);
       final MySqlCdcConnectorMetadataInjector mySqlCdcConnectorMetadataInjector = new MySqlCdcConnectorMetadataInjector();

airbyte-integrations/connectors/source-postgres-strict-encrypt/Dockerfile

@@ -24,5 +24,5 @@ ENV APPLICATION source-postgres-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=2.0.29
+LABEL io.airbyte.version=2.0.30
 LABEL io.airbyte.name=airbyte/source-postgres-strict-encrypt

airbyte-integrations/connectors/source-postgres-strict-encrypt/metadata.yaml

@@ -12,7 +12,7 @@ data:
   connectorType: source
   definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
   maxSecondsBetweenMessages: 7200
-  dockerImageTag: 2.0.29
+  dockerImageTag: 2.0.30
   dockerRepository: airbyte/source-postgres-strict-encrypt
   githubIssueLabel: source-postgres
   icon: postgresql.svg

airbyte-integrations/connectors/source-postgres-strict-encrypt/src/test-integration/resources/expected_strict_encrypt_spec.json

@@ -281,6 +281,15 @@
                 "min": 120,
                 "max": 1200
               },
+              "queue_size": {
+                "type": "integer",
+                "title": "Size of the queue (Advanced)",
+                "description": "The size of the internal queue. This may interfere with memory consumption and efficiency of the connector, please be careful.",
+                "default": 10000,
+                "order": 6,
+                "min": 1000,
+                "max": 10000
+              },
               "lsn_commit_behaviour": {
                 "type": "string",
                 "title": "LSN commit behaviour",
@@ -290,7 +299,7 @@
                   "After loading Data in the destination"
                 ],
                 "default": "After loading Data in the destination",
-                "order": 6
+                "order": 7
               }
             }
           }

airbyte-integrations/connectors/source-postgres/Dockerfile

@@ -24,5 +24,5 @@ ENV APPLICATION source-postgres
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=2.0.29
+LABEL io.airbyte.version=2.0.30
 LABEL io.airbyte.name=airbyte/source-postgres

airbyte-integrations/connectors/source-postgres/metadata.yaml

@@ -6,7 +6,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
-  dockerImageTag: 2.0.29
+  dockerImageTag: 2.0.30
   maxSecondsBetweenMessages: 7200
   dockerRepository: airbyte/source-postgres
   githubIssueLabel: source-postgres

airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresCdcProperties.java

@@ -62,7 +62,7 @@ private static Properties commonProperties(final JdbcDatabase database) {
     }
 
     // Check params for SSL connection in config and add properties for CDC SSL connection
-    // https://debezium.io/documentation/reference/2.1/connectors/postgresql.html#postgresql-property-database-sslmode
+    // https://debezium.io/documentation/reference/2.2/connectors/postgresql.html#postgresql-property-database-sslmode
     if (!sourceConfig.has(JdbcUtils.SSL_KEY) || sourceConfig.get(JdbcUtils.SSL_KEY).asBoolean()) {
       if (sourceConfig.has(JdbcUtils.SSL_MODE_KEY) && sourceConfig.get(JdbcUtils.SSL_MODE_KEY).has(JdbcUtils.MODE_KEY)) {