forked from debezium/debezium
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
4e55ebd
commit 825a5b6
Showing
20 changed files
with
739 additions
and
22 deletions.
There are no files selected for viewing
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
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
110 changes: 110 additions & 0 deletions
110
...ain/java/io/debezium/connector/postgresql/metrics/AbstractYugabyteDBPartitionMetrics.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,110 @@ | ||
| package io.debezium.connector.postgresql.metrics; | ||
|
|
||
| import io.debezium.connector.common.CdcSourceTaskContext; | ||
| import io.debezium.connector.postgresql.PostgresTaskContext; | ||
| import io.debezium.data.Envelope; | ||
| import io.debezium.metrics.Metrics; | ||
| import io.debezium.pipeline.ConnectorEvent; | ||
| import io.debezium.pipeline.meters.CommonEventMeter; | ||
| import io.debezium.pipeline.source.spi.EventMetadataProvider; | ||
| import io.debezium.pipeline.spi.OffsetContext; | ||
| import io.debezium.spi.schema.DataCollectionId; | ||
| import org.apache.kafka.connect.data.Struct; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| abstract class AbstractYugabyteDBPartitionMetrics extends YugabyteDBMetrics implements YugabyteDBPartitionMetricsMXBean { | ||
| private final CommonEventMeter commonEventMeter; | ||
|
|
||
| AbstractYugabyteDBPartitionMetrics(CdcSourceTaskContext taskContext, Map<String, String> tags, | ||
| EventMetadataProvider metadataProvider) { | ||
| super(taskContext, tags); | ||
| this.commonEventMeter = new CommonEventMeter(taskContext.getClock(), metadataProvider); | ||
| } | ||
|
|
||
| @Override | ||
| public String getLastEvent() { | ||
| return commonEventMeter.getLastEvent(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getMilliSecondsSinceLastEvent() { | ||
| return commonEventMeter.getMilliSecondsSinceLastEvent(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getTotalNumberOfEventsSeen() { | ||
| return commonEventMeter.getTotalNumberOfEventsSeen(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getTotalNumberOfCreateEventsSeen() { | ||
| return commonEventMeter.getTotalNumberOfCreateEventsSeen(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getTotalNumberOfUpdateEventsSeen() { | ||
| return commonEventMeter.getTotalNumberOfUpdateEventsSeen(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getTotalNumberOfDeleteEventsSeen() { | ||
| return commonEventMeter.getTotalNumberOfDeleteEventsSeen(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getNumberOfEventsFiltered() { | ||
| return commonEventMeter.getNumberOfEventsFiltered(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getNumberOfErroneousEvents() { | ||
| return commonEventMeter.getNumberOfErroneousEvents(); | ||
| } | ||
|
|
||
| /** | ||
| * Invoked if an event is processed for a captured table. | ||
| */ | ||
| void onEvent(DataCollectionId source, OffsetContext offset, Object key, Struct value, Envelope.Operation operation) { | ||
| commonEventMeter.onEvent(source, offset, key, value, operation); | ||
| } | ||
|
|
||
| /** | ||
| * Invoked for events pertaining to non-captured tables. | ||
| */ | ||
| void onFilteredEvent(String event) { | ||
| commonEventMeter.onFilteredEvent(); | ||
| } | ||
|
|
||
| /** | ||
| * Invoked for events pertaining to non-captured tables. | ||
| */ | ||
| void onFilteredEvent(String event, Envelope.Operation operation) { | ||
| commonEventMeter.onFilteredEvent(operation); | ||
| } | ||
|
|
||
| /** | ||
| * Invoked for events that cannot be processed. | ||
| */ | ||
| void onErroneousEvent(String event) { | ||
| commonEventMeter.onErroneousEvent(); | ||
| } | ||
|
|
||
| /** | ||
| * Invoked for events that cannot be processed. | ||
| */ | ||
| void onErroneousEvent(String event, Envelope.Operation operation) { | ||
| commonEventMeter.onErroneousEvent(operation); | ||
| } | ||
|
|
||
| /** | ||
| * Invoked for events that represent a connector event. | ||
| */ | ||
| void onConnectorEvent(ConnectorEvent event) { | ||
| } | ||
|
|
||
| @Override | ||
| public void reset() { | ||
| commonEventMeter.reset(); | ||
| } | ||
| } |
118 changes: 118 additions & 0 deletions
118
...src/main/java/io/debezium/connector/postgresql/metrics/AbstractYugabyteDBTaskMetrics.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,118 @@ | ||
| package io.debezium.connector.postgresql.metrics; | ||
|
|
||
| import io.debezium.connector.base.ChangeEventQueueMetrics; | ||
| import io.debezium.connector.common.CdcSourceTaskContext; | ||
| import io.debezium.connector.postgresql.PostgresPartition; | ||
| import io.debezium.data.Envelope; | ||
| import io.debezium.metrics.Metrics; | ||
| import io.debezium.pipeline.ConnectorEvent; | ||
| import io.debezium.pipeline.metrics.ChangeEventSourceMetrics; | ||
| import io.debezium.pipeline.spi.OffsetContext; | ||
| import io.debezium.spi.schema.DataCollectionId; | ||
| import io.debezium.util.Collect; | ||
| import org.apache.kafka.connect.data.Struct; | ||
|
|
||
| import java.util.Collection; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.function.Consumer; | ||
| import java.util.function.Function; | ||
|
|
||
| abstract class AbstractYugabyteDBTaskMetrics<B extends AbstractYugabyteDBPartitionMetrics> extends YugabyteDBMetrics | ||
| implements ChangeEventSourceMetrics<PostgresPartition>, YugabyteDBTaskMetricsMXBean { | ||
|
|
||
| private final ChangeEventQueueMetrics changeEventQueueMetrics; | ||
| private final Map<PostgresPartition, B> beans = new HashMap<>(); | ||
|
|
||
| AbstractYugabyteDBTaskMetrics(CdcSourceTaskContext taskContext, | ||
| String contextName, | ||
| ChangeEventQueueMetrics changeEventQueueMetrics, | ||
| Collection<PostgresPartition> partitions, | ||
| Function<PostgresPartition, B> beanFactory) { | ||
| super(taskContext, Collect.linkMapOf( | ||
| "server", taskContext.getConnectorName(), | ||
| "task", taskContext.getTaskId(), | ||
| "context", contextName)); | ||
| this.changeEventQueueMetrics = changeEventQueueMetrics; | ||
|
|
||
| for (PostgresPartition partition : partitions) { | ||
| beans.put(partition, beanFactory.apply(partition)); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized void register() { | ||
| super.register(); | ||
| beans.values().forEach(YugabyteDBMetrics::register); | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized void unregister() { | ||
| beans.values().forEach(YugabyteDBMetrics::unregister); | ||
| super.unregister(); | ||
| } | ||
|
|
||
| @Override | ||
| public void reset() { | ||
| beans.values().forEach(B::reset); | ||
| } | ||
|
|
||
| @Override | ||
| public void onEvent(PostgresPartition partition, DataCollectionId source, OffsetContext offset, Object key, | ||
| Struct value, Envelope.Operation operation) { | ||
| onPartitionEvent(partition, bean -> bean.onEvent(source, offset, key, value, operation)); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFilteredEvent(PostgresPartition partition, String event) { | ||
| onPartitionEvent(partition, bean -> bean.onFilteredEvent(event)); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFilteredEvent(PostgresPartition partition, String event, Envelope.Operation operation) { | ||
| onPartitionEvent(partition, bean -> bean.onFilteredEvent(event, operation)); | ||
| } | ||
|
|
||
| @Override | ||
| public void onErroneousEvent(PostgresPartition partition, String event) { | ||
| onPartitionEvent(partition, bean -> bean.onErroneousEvent(event)); | ||
| } | ||
|
|
||
| @Override | ||
| public void onErroneousEvent(PostgresPartition partition, String event, Envelope.Operation operation) { | ||
| onPartitionEvent(partition, bean -> bean.onErroneousEvent(event, operation)); | ||
| } | ||
|
|
||
| @Override | ||
| public void onConnectorEvent(PostgresPartition partition, ConnectorEvent event) { | ||
| onPartitionEvent(partition, bean -> bean.onConnectorEvent(event)); | ||
| } | ||
|
|
||
| @Override | ||
| public int getQueueTotalCapacity() { | ||
| return changeEventQueueMetrics.totalCapacity(); | ||
| } | ||
|
|
||
| @Override | ||
| public int getQueueRemainingCapacity() { | ||
| return changeEventQueueMetrics.remainingCapacity(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getMaxQueueSizeInBytes() { | ||
| return changeEventQueueMetrics.maxQueueSizeInBytes(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getCurrentQueueSizeInBytes() { | ||
| return changeEventQueueMetrics.currentQueueSizeInBytes(); | ||
| } | ||
|
|
||
| protected void onPartitionEvent(PostgresPartition partition, Consumer<B> handler) { | ||
| B bean = beans.get(partition); | ||
| if (bean == null) { | ||
| throw new IllegalArgumentException("MBean for partition " + partition + " are not registered"); | ||
| } | ||
| handler.accept(bean); | ||
| } | ||
| } |
Oops, something went wrong.