-
Notifications
You must be signed in to change notification settings - Fork 35
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add batch size wait to optimize batch size (#16)
* rework batch wait
- Loading branch information
1 parent
da19218
commit e82f3b7
Showing
23 changed files
with
540 additions
and
85 deletions.
There are no files selected for viewing
92 changes: 92 additions & 0 deletions
92
debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/DebeziumMetrics.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| /* | ||
| * | ||
| * * Copyright memiiso Authors. | ||
| * * | ||
| * * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| */ | ||
|
|
||
| package io.debezium.server.iceberg; | ||
|
|
||
| import io.debezium.DebeziumException; | ||
| import io.debezium.config.CommonConnectorConfig; | ||
|
|
||
| import java.lang.management.ManagementFactory; | ||
| import java.util.Optional; | ||
| import javax.enterprise.context.Dependent; | ||
| import javax.management.MBeanServer; | ||
| import javax.management.ObjectName; | ||
|
|
||
| import org.eclipse.microprofile.config.inject.ConfigProperty; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** | ||
| * Optimizes batch size around 85%-90% of max,batch.size using dynamically calculated sleep(ms) | ||
| * | ||
| * @author Ismail Simsek | ||
| */ | ||
| @Dependent | ||
| public class DebeziumMetrics { | ||
| protected static final Logger LOGGER = LoggerFactory.getLogger(DebeziumMetrics.class); | ||
| final MBeanServer mbeanServer = ManagementFactory.getPlatformMBeanServer(); | ||
| @ConfigProperty(name = "debezium.sink.batch.metrics.snapshot-mbean", defaultValue = "") | ||
| Optional<String> snapshotMbean; | ||
| @ConfigProperty(name = "debezium.sink.batch.metrics.streaming-mbean", defaultValue = "") | ||
| Optional<String> streamingMbean; | ||
| @ConfigProperty(name = "debezium.source.max.queue.size", defaultValue = CommonConnectorConfig.DEFAULT_MAX_QUEUE_SIZE + "") | ||
| int maxQueueSize; | ||
|
|
||
| ObjectName snapshotMetricsObjectName; | ||
| ObjectName streamingMetricsObjectName; | ||
|
|
||
| public void initizalize() throws DebeziumException { | ||
| assert snapshotMbean.isPresent() : | ||
| "Snapshot metrics Mbean `debezium.sink.batch.metrics.snapshot-mbean` not provided"; | ||
| assert streamingMbean.isPresent() : | ||
| "Streaming metrics Mbean `debezium.sink.batch.metrics.streaming-mbean` not provided"; | ||
| try { | ||
| snapshotMetricsObjectName = new ObjectName(snapshotMbean.get()); | ||
| streamingMetricsObjectName = new ObjectName(streamingMbean.get()); | ||
| } catch (Exception e) { | ||
| throw new DebeziumException(e); | ||
| } | ||
| } | ||
|
|
||
| public boolean snapshotRunning() { | ||
| try { | ||
| return (boolean) mbeanServer.getAttribute(snapshotMetricsObjectName, "SnapshotRunning"); | ||
| } catch (Exception e) { | ||
| throw new DebeziumException(e); | ||
| } | ||
| } | ||
|
|
||
| public boolean snapshotCompleted() { | ||
| try { | ||
| return (boolean) mbeanServer.getAttribute(snapshotMetricsObjectName, "SnapshotCompleted"); | ||
| } catch (Exception e) { | ||
| throw new DebeziumException(e); | ||
| } | ||
| } | ||
|
|
||
| public int streamingQueueRemainingCapacity() { | ||
| try { | ||
| return (int) mbeanServer.getAttribute(streamingMetricsObjectName, "QueueRemainingCapacity"); | ||
| } catch (Exception e) { | ||
| throw new DebeziumException(e); | ||
| } | ||
| } | ||
|
|
||
| public int streamingQueueCurrentSize() { | ||
| return maxQueueSize - streamingQueueRemainingCapacity(); | ||
| } | ||
|
|
||
| public long streamingMilliSecondsBehindSource() { | ||
| try { | ||
| return (long) mbeanServer.getAttribute(streamingMetricsObjectName, "MilliSecondsBehindSource"); | ||
| } catch (Exception e) { | ||
| throw new DebeziumException(e); | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
23 changes: 23 additions & 0 deletions
23
...g-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/InterfaceBatchSizeWait.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| /* | ||
| * | ||
| * * Copyright memiiso Authors. | ||
| * * | ||
| * * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| */ | ||
|
|
||
| package io.debezium.server.iceberg.batchsizewait; | ||
|
|
||
| /** | ||
| * Implementation of the consumer that delivers the messages into Amazon S3 destination. | ||
| * | ||
| * @author Ismail Simsek | ||
| */ | ||
| public interface InterfaceBatchSizeWait { | ||
|
|
||
| default void initizalize() { | ||
| } | ||
|
|
||
| void waitMs(Integer numRecordsProcessed, Integer processingTimeMs) throws InterruptedException; | ||
|
|
||
| } |
Oops, something went wrong.