DBZ-2662 Handle ChangeEventQueue by the size in bytes

usabilla · Nov 16, 2020 · 17a424d · 17a424d
1 parent 4faf13d
commit 17a424d
Show file tree

Hide file tree

Showing 18 changed files with 700 additions and 3 deletions.
diff --git a/...-assembly-descriptors/src/main/resources/assemblies/connector-distribution-no-drivers.xml b/...-assembly-descriptors/src/main/resources/assemblies/connector-distribution-no-drivers.xml
@@ -23,6 +23,9 @@
               <exclude>com.fasterxml.jackson.datatype:jackson-datatype-jsr310:*</exclude>
               <exclude>org.reflections:reflections:*</exclude>
 
+              <!-- Exclude guava dependencies -->
+              <exclude>com.google.guava:listenablefuture:*</exclude>
+
               <!-- Exclude connection drivers -->
 	      <exclude>${assembly.exclude.1}</exclude>
 	      <exclude>${assembly.exclude.2}</exclude>

diff --git a/debezium-assembly-descriptors/src/main/resources/assemblies/connector-distribution.xml b/debezium-assembly-descriptors/src/main/resources/assemblies/connector-distribution.xml
@@ -22,6 +22,9 @@
               <exclude>com.fasterxml.jackson.core:jackson-annotations:*</exclude>
               <exclude>com.fasterxml.jackson.datatype:jackson-datatype-jsr310:*</exclude>
               <exclude>org.reflections:reflections:*</exclude>
+
+              <!-- Exclude guava dependencies -->
+              <exclude>com.google.guava:listenablefuture:*</exclude>
           </excludes>
       </dependencySet>
       <dependencySet>

diff --git a/...m-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnectorTask.java b/...m-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnectorTask.java
@@ -85,6 +85,7 @@ public ChangeEventSourceCoordinator start(Configuration config) {
                     .pollInterval(connectorConfig.getPollInterval())
                     .maxBatchSize(connectorConfig.getMaxBatchSize())
                     .maxQueueSize(connectorConfig.getMaxQueueSize())
+                    .maxQueueSizeInBytes(connectorConfig.getMaxQueueSizeInBytes())
                     .loggingContextSupplier(() -> taskContext.configureLoggingContext(CONTEXT_NAME))
                     .build();
 

diff --git a/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/AbstractReader.java b/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/AbstractReader.java
@@ -84,6 +84,17 @@ public int totalCapacity() {
             public int remainingCapacity() {
                 return records.remainingCapacity();
             }
+
+            @Override
+            public long maxQueueSizeInBytes() {
+                return context.getConnectorConfig().getMaxQueueSizeInBytes();
+            }
+
+            @Override
+            public long currentQueueSizeInBytes() {
+                // return 0 since MySQL connector doesn't use ChangeEventQueue implementation
+                return 0;
+            }
         };
     }
 

diff --git a/...nector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresConnectorTask.java b/...nector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresConnectorTask.java
@@ -150,6 +150,7 @@ public ChangeEventSourceCoordinator start(Configuration config) {
                     .pollInterval(connectorConfig.getPollInterval())
                     .maxBatchSize(connectorConfig.getMaxBatchSize())
                     .maxQueueSize(connectorConfig.getMaxQueueSize())
+                    .maxQueueSizeInBytes(connectorConfig.getMaxQueueSizeInBytes())
                     .loggingContextSupplier(() -> taskContext.configureLoggingContext(CONTEXT_NAME))
                     .build();
 

diff --git a/...-connector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresMetricsIT.java b/...-connector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresMetricsIT.java
@@ -21,6 +21,7 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import io.debezium.config.Configuration;
 import io.debezium.connector.postgresql.PostgresConnectorConfig.SnapshotMode;
 
 /**
@@ -190,6 +191,97 @@ private void assertStreamingMetrics() throws Exception {
         // Assertions.assertThat(mBeanServer.getAttribute(getStreamingMetricsObjectName(), "MonitoredTables")).isEqualTo(new String[] {"public.simple"});
     }
 
+    @Test
+    public void twoRecordsInQueue() throws Exception {
+        final MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
+        TestHelper.execute(INIT_STATEMENTS, INSERT_STATEMENTS);
+        final int recordCount = 2;
+
+        Configuration.Builder configBuilder = TestHelper.defaultConfig()
+                .with(PostgresConnectorConfig.SNAPSHOT_MODE, SnapshotMode.NEVER)
+                .with(PostgresConnectorConfig.DROP_SLOT_ON_STOP, Boolean.TRUE)
+                .with(PostgresConnectorConfig.MAX_QUEUE_SIZE, 10)
+                .with(PostgresConnectorConfig.MAX_BATCH_SIZE, 1)
+                .with(PostgresConnectorConfig.POLL_INTERVAL_MS, 5000L)
+                .with(PostgresConnectorConfig.MAX_QUEUE_SIZE_IN_BYTES, 10000L);
+        start(PostgresConnector.class, configBuilder.build());
+
+        waitForStreamingToStart();
+        for (int i = 0; i < recordCount - 1; i++) {
+            TestHelper.execute(INSERT_STATEMENTS);
+        }
+        Awaitility.await()
+                .alias("MBean attribute was not an expected value")
+                .pollInterval(100, TimeUnit.MILLISECONDS)
+                .atMost(TestHelper.waitTimeForRecords(), TimeUnit.SECONDS)
+                .ignoreException(InstanceNotFoundException.class)
+                .until(() -> {
+                    long value = (long) mBeanServer.getAttribute(getStreamingMetricsObjectName(), "CurrentQueueSizeInBytes");
+                    return value > 0;
+                });
+        Assertions.assertThat(mBeanServer.getAttribute(getStreamingMetricsObjectName(), "CurrentQueueSizeInBytes")).isNotEqualTo(0L);
+        Awaitility.await()
+                .alias("MBean attribute was not an expected value")
+                .pollInterval(100, TimeUnit.MILLISECONDS)
+                .atMost(TestHelper.waitTimeForRecords(), TimeUnit.SECONDS)
+                .ignoreException(InstanceNotFoundException.class)
+                .until(() -> {
+                    int value = (int) mBeanServer.getAttribute(getStreamingMetricsObjectName(), "QueueRemainingCapacity");
+                    return value == 8;
+                });
+        Assertions.assertThat(mBeanServer.getAttribute(getStreamingMetricsObjectName(), "QueueRemainingCapacity")).isEqualTo(8);
+
+        SourceRecords records = consumeRecordsByTopic(recordCount);
+
+        Assertions.assertThat(mBeanServer.getAttribute(getStreamingMetricsObjectName(), "CurrentQueueSizeInBytes")).isEqualTo(0L);
+        stopConnector();
+    }
+
+    @Test
+    public void oneRecordInQueue() throws Exception {
+        final MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
+        TestHelper.execute(INIT_STATEMENTS, INSERT_STATEMENTS);
+        final int recordCount = 2;
+
+        Configuration.Builder configBuilder = TestHelper.defaultConfig()
+                .with(PostgresConnectorConfig.SNAPSHOT_MODE, SnapshotMode.NEVER)
+                .with(PostgresConnectorConfig.DROP_SLOT_ON_STOP, Boolean.TRUE)
+                .with(PostgresConnectorConfig.MAX_QUEUE_SIZE, 10)
+                .with(PostgresConnectorConfig.MAX_BATCH_SIZE, 1)
+                .with(PostgresConnectorConfig.POLL_INTERVAL_MS, 5000L)
+                .with(PostgresConnectorConfig.MAX_QUEUE_SIZE_IN_BYTES, 10L);
+        start(PostgresConnector.class, configBuilder.build());
+
+        waitForStreamingToStart();
+        for (int i = 0; i < recordCount - 1; i++) {
+            TestHelper.execute(INSERT_STATEMENTS);
+        }
+        Awaitility.await()
+                .alias("MBean attribute was not an expected value")
+                .pollInterval(100, TimeUnit.MILLISECONDS)
+                .atMost(TestHelper.waitTimeForRecords(), TimeUnit.SECONDS)
+                .ignoreException(InstanceNotFoundException.class)
+                .until(() -> {
+                    long value = (long) mBeanServer.getAttribute(getStreamingMetricsObjectName(), "CurrentQueueSizeInBytes");
+                    return value > 0;
+                });
+        Assertions.assertThat(mBeanServer.getAttribute(getStreamingMetricsObjectName(), "CurrentQueueSizeInBytes")).isNotEqualTo(0L);
+        Awaitility.await()
+                .alias("MBean attribute was not an expected value")
+                .pollInterval(100, TimeUnit.MILLISECONDS)
+                .atMost(TestHelper.waitTimeForRecords(), TimeUnit.SECONDS)
+                .ignoreException(InstanceNotFoundException.class)
+                .until(() -> {
+                    int value = (int) mBeanServer.getAttribute(getStreamingMetricsObjectName(), "QueueRemainingCapacity");
+                    return value == 9;
+                });
+        Assertions.assertThat(mBeanServer.getAttribute(getStreamingMetricsObjectName(), "QueueRemainingCapacity")).isEqualTo(9);
+
+        SourceRecords records = consumeRecordsByTopic(recordCount);
+        Assertions.assertThat(mBeanServer.getAttribute(getStreamingMetricsObjectName(), "CurrentQueueSizeInBytes")).isEqualTo(0L);
+        stopConnector();
+    }
+
     private ObjectName getSnapshotMetricsObjectName() throws MalformedObjectNameException {
         return getSnapshotMetricsObjectName("postgres", TestHelper.TEST_SERVER);
     }

diff --git a/...ector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnectorTask.java b/...ector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnectorTask.java
@@ -95,6 +95,7 @@ public ChangeEventSourceCoordinator start(Configuration config) {
                 .pollInterval(connectorConfig.getPollInterval())
                 .maxBatchSize(connectorConfig.getMaxBatchSize())
                 .maxQueueSize(connectorConfig.getMaxQueueSize())
+                .maxQueueSizeInBytes(connectorConfig.getMaxQueueSizeInBytes())
                 .loggingContextSupplier(() -> taskContext.configureLoggingContext(CONTEXT_NAME))
                 .build();
 

diff --git a/debezium-core/pom.xml b/debezium-core/pom.xml
@@ -101,6 +101,30 @@
             <artifactId>kafka-connect-avro-converter</artifactId>
             <scope>test</scope>
         </dependency>
+        <!-- open libraries -->
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>29.0-jre</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.code.findbugs</groupId>
+                    <artifactId>jsr305</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.google.errorprone</groupId>
+                    <artifactId>error_prone_annotations</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.google.j2objc</groupId>
+                    <artifactId>j2objc-annotations</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.checkerframework</groupId>
+                    <artifactId>checker-qual</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
     </dependencies>
     <build>
         <resources>

diff --git a/debezium-core/src/main/java/io/debezium/config/CommonConnectorConfig.java b/debezium-core/src/main/java/io/debezium/config/CommonConnectorConfig.java
@@ -242,6 +242,7 @@ public static BinaryHandlingMode parse(String value, String defaultValue) {
     public static final String DATABASE_CONFIG_PREFIX = "database.";
     private static final String CONVERTER_TYPE_SUFFIX = ".type";
     public static final long DEFAULT_RETRIABLE_RESTART_WAIT = 10000L;
+    public static final long DEFAULT_MAX_QUEUE_SIZE_IN_BYTES = 0; // In case we don't want to pass max.queue.size.in.bytes;
 
     public static final Field RETRIABLE_RESTART_WAIT = Field.create("retriable.restart.connector.wait.ms")
             .withDisplayName("Retriable restart wait (ms)")
@@ -294,6 +295,16 @@ public static BinaryHandlingMode parse(String value, String defaultValue) {
             .withDefault(DEFAULT_POLL_INTERVAL_MILLIS)
             .withValidation(Field::isPositiveInteger);
 
+    public static final Field MAX_QUEUE_SIZE_IN_BYTES = Field.create("max.queue.size.in.bytes")
+            .withDisplayName("Change event buffer size in bytes")
+            .withType(Type.LONG)
+            .withWidth(Width.LONG)
+            .withImportance(Importance.MEDIUM)
+            .withDescription("Maximum size of the queue in bytes for change events read from the database log but not yet recorded or forwarded. Defaults to "
+                    + DEFAULT_MAX_QUEUE_SIZE_IN_BYTES + ". Mean the feature is not enabled")
+            .withDefault(DEFAULT_MAX_QUEUE_SIZE_IN_BYTES)
+            .withValidation(Field::isNonNegativeLong);
+
     public static final Field SNAPSHOT_DELAY_MS = Field.create("snapshot.delay.ms")
             .withDisplayName("Snapshot Delay (milliseconds)")
             .withType(Type.LONG)
@@ -405,6 +416,7 @@ public static BinaryHandlingMode parse(String value, String defaultValue) {
                     MAX_BATCH_SIZE,
                     MAX_QUEUE_SIZE,
                     POLL_INTERVAL_MS,
+                    MAX_QUEUE_SIZE_IN_BYTES,
                     PROVIDE_TRANSACTION_METADATA,
                     SKIPPED_OPERATIONS,
                     SNAPSHOT_DELAY_MS,
@@ -426,6 +438,7 @@ public static BinaryHandlingMode parse(String value, String defaultValue) {
     private final boolean emitTombstoneOnDelete;
     private final int maxQueueSize;
     private final int maxBatchSize;
+    private final long maxQueueSizeInBytes;
     private final Duration pollInterval;
     private final String logicalName;
     private final String heartbeatTopicsPrefix;
@@ -447,6 +460,7 @@ protected CommonConnectorConfig(Configuration config, String logicalName, int de
         this.maxQueueSize = config.getInteger(MAX_QUEUE_SIZE);
         this.maxBatchSize = config.getInteger(MAX_BATCH_SIZE);
         this.pollInterval = config.getDuration(POLL_INTERVAL_MS, ChronoUnit.MILLIS);
+        this.maxQueueSizeInBytes = config.getLong(MAX_QUEUE_SIZE_IN_BYTES);
         this.logicalName = logicalName;
         this.heartbeatTopicsPrefix = config.getString(Heartbeat.HEARTBEAT_TOPICS_PREFIX);
         this.snapshotDelayMs = Duration.ofMillis(config.getLong(SNAPSHOT_DELAY_MS));
@@ -484,6 +498,10 @@ public int getMaxBatchSize() {
         return maxBatchSize;
     }
 
+    public long getMaxQueueSizeInBytes() {
+        return maxQueueSizeInBytes;
+    }
+
     public Duration getPollInterval() {
         return pollInterval;
     }

diff --git a/debezium-core/src/main/java/io/debezium/connector/base/ChangeEventQueue.java b/debezium-core/src/main/java/io/debezium/connector/base/ChangeEventQueue.java
@@ -8,8 +8,11 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 
 import org.apache.kafka.connect.source.SourceRecord;
@@ -22,6 +25,7 @@
 import io.debezium.util.LoggingContext;
 import io.debezium.util.LoggingContext.PreviousContext;
 import io.debezium.util.Metronome;
+import io.debezium.util.ObjectSizeCalculator;
 import io.debezium.util.Threads;
 import io.debezium.util.Threads.Timer;
 
@@ -57,19 +61,24 @@ public class ChangeEventQueue<T> implements ChangeEventQueueMetrics {
     private final Duration pollInterval;
     private final int maxBatchSize;
     private final int maxQueueSize;
+    private final long maxQueueSizeInBytes;
     private final BlockingQueue<T> queue;
     private final Metronome metronome;
     private final Supplier<PreviousContext> loggingContextSupplier;
+    private AtomicLong currentQueueSizeInBytes = new AtomicLong(0);
+    private Map<T, Long> objectMap = new ConcurrentHashMap<>();
 
     private volatile RuntimeException producerException;
 
-    private ChangeEventQueue(Duration pollInterval, int maxQueueSize, int maxBatchSize, Supplier<LoggingContext.PreviousContext> loggingContextSupplier) {
+    private ChangeEventQueue(Duration pollInterval, int maxQueueSize, int maxBatchSize, Supplier<LoggingContext.PreviousContext> loggingContextSupplier,
+                             long maxQueueSizeInBytes) {
         this.pollInterval = pollInterval;
         this.maxBatchSize = maxBatchSize;
         this.maxQueueSize = maxQueueSize;
         this.queue = new LinkedBlockingDeque<>(maxQueueSize);
         this.metronome = Metronome.sleeper(pollInterval, Clock.SYSTEM);
         this.loggingContextSupplier = loggingContextSupplier;
+        this.maxQueueSizeInBytes = maxQueueSizeInBytes;
     }
 
     public static class Builder<T> {
@@ -78,6 +87,7 @@ public static class Builder<T> {
         private int maxQueueSize;
         private int maxBatchSize;
         private Supplier<LoggingContext.PreviousContext> loggingContextSupplier;
+        private long maxQueueSizeInBytes;
 
         public Builder<T> pollInterval(Duration pollInterval) {
             this.pollInterval = pollInterval;
@@ -99,8 +109,13 @@ public Builder<T> loggingContextSupplier(Supplier<LoggingContext.PreviousContext
             return this;
         }
 
+        public Builder<T> maxQueueSizeInBytes(long maxQueueSizeInBytes) {
+            this.maxQueueSizeInBytes = maxQueueSizeInBytes;
+            return this;
+        }
+
         public ChangeEventQueue<T> build() {
-            return new ChangeEventQueue<T>(pollInterval, maxQueueSize, maxBatchSize, loggingContextSupplier);
+            return new ChangeEventQueue<T>(pollInterval, maxQueueSize, maxBatchSize, loggingContextSupplier, maxQueueSizeInBytes);
         }
     }
 
@@ -126,9 +141,18 @@ public void enqueue(T record) throws InterruptedException {
         if (LOGGER.isDebugEnabled()) {
             LOGGER.debug("Enqueuing source record '{}'", record);
         }
-
+        // Waiting for queue to add more record.
+        while (maxQueueSizeInBytes > 0 && currentQueueSizeInBytes.get() > maxQueueSizeInBytes) {
+            Thread.sleep(pollInterval.toMillis());
+        }
         // this will also raise an InterruptedException if the thread is interrupted while waiting for space in the queue
         queue.put(record);
+        // If we pass a positiveLong max.queue.size.in.bytes to enable handling queue size in bytes feature
+        if (maxQueueSizeInBytes > 0) {
+            long messageSize = ObjectSizeCalculator.getObjectSize(record);
+            objectMap.put(record, messageSize);
+            currentQueueSizeInBytes.addAndGet(messageSize);
+        }
     }
 
     /**
@@ -154,6 +178,14 @@ public List<T> poll() throws InterruptedException {
                 metronome.pause();
                 LOGGER.debug("checking for more records...");
             }
+            if (maxQueueSizeInBytes > 0 && records.size() > 0) {
+                records.parallelStream().forEach((record) -> {
+                    if (objectMap.containsKey(record)) {
+                        currentQueueSizeInBytes.addAndGet(-objectMap.get(record));
+                        objectMap.remove(record);
+                    }
+                });
+            }
             return records;
         }
         finally {
@@ -180,4 +212,14 @@ public int totalCapacity() {
     public int remainingCapacity() {
         return queue.remainingCapacity();
     }
+
+    @Override
+    public long maxQueueSizeInBytes() {
+        return maxQueueSizeInBytes;
+    }
+
+    @Override
+    public long currentQueueSizeInBytes() {
+        return currentQueueSizeInBytes.get();
+    }
 }
diff --git a/debezium-core/src/main/java/io/debezium/connector/base/ChangeEventQueueMetrics.java b/debezium-core/src/main/java/io/debezium/connector/base/ChangeEventQueueMetrics.java
@@ -10,4 +10,8 @@ public interface ChangeEventQueueMetrics {
     int totalCapacity();
 
     int remainingCapacity();
+
+    long maxQueueSizeInBytes();
+
+    long currentQueueSizeInBytes();
 }
diff --git a/debezium-core/src/main/java/io/debezium/pipeline/metrics/ChangeEventSourceMetricsMXBean.java b/debezium-core/src/main/java/io/debezium/pipeline/metrics/ChangeEventSourceMetricsMXBean.java
@@ -28,5 +28,9 @@ public interface ChangeEventSourceMetricsMXBean {
 
     int getQueueRemainingCapacity();
 
+    long getMaxQueueSizeInBytes();
+
+    long getCurrentQueueSizeInBytes();
+
     void reset();
 }