alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-7601 Support TRUNCATE event

Browse Source
This commit is contained in:
mfvitale 2024-08-13 14:40:03 +02:00 committed by Jiri Pechanec
parent 1f0a61fb30
commit 6977cb47a1
5 changed files with 29 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
debezium-connector-sqlserver/src/main/java/io/debezium/connector/sqlserver/metrics/AbstractSqlServerPartitionMetrics.java
View File

@ -91,6 +91,11 @@ public Map<String, Long> getNumberOfUpdateEventsSeen() {
return activityMonitoringMeter.getNumberOfUpdateEventsSeen();
}
@Override
public Map<String, Long> getNumberOfTruncateEventsSeen() {
return activityMonitoringMeter.getNumberOfTruncateEventsSeen();
}
@Override
public void pause() {
activityMonitoringMeter.pause();

5
debezium-core/src/main/java/io/debezium/pipeline/metrics/DefaultStreamingChangeEventSourceMetrics.java
View File

@ -130,4 +130,9 @@ public Map<String, Long> getNumberOfDeleteEventsSeen() {
public Map<String, Long> getNumberOfUpdateEventsSeen() {
return activityMonitoringMeter.getNumberOfUpdateEventsSeen();
}
@Override
public Map<String, Long> getNumberOfTruncateEventsSeen() {
return activityMonitoringMeter.getNumberOfTruncateEventsSeen();
}
}

2
debezium-core/src/main/java/io/debezium/transforms/ActivityMonitoringMXBean.java
View File

@ -18,6 +18,8 @@ public interface ActivityMonitoringMXBean {
Map<String, Long> getNumberOfUpdateEventsSeen();
Map<String, Long> getNumberOfTruncateEventsSeen();
void pause();
void resume();

12
debezium-core/src/main/java/io/debezium/transforms/ActivityMonitoringMeter.java
View File

@ -26,6 +26,7 @@ public class ActivityMonitoringMeter implements ActivityMonitoringMXBean {
private final ActivityCounter createCount = new ActivityCounter();
private final ActivityCounter updateCount = new ActivityCounter();
private final ActivityCounter deleteCount = new ActivityCounter();
private final ActivityCounter truncateCount = new ActivityCounter();
private boolean isPaused = false;
@ -48,11 +49,14 @@ public void onEvent(DataCollectionId source, OffsetContext offset, Object key, S
case DELETE:
deleteCount.add(1, tableName);
break;
case TRUNCATE:
truncateCount.add(1, tableName);
break;
default:
break;
}
LOGGER.trace("Counter status create:{}, delete:{}, update:{}", createCount, deleteCount, updateCount);
LOGGER.trace("Counter status create:{}, delete:{}, update:{}, truncate:{}", createCount, deleteCount, updateCount, truncateCount);
}
@Override
@ -70,6 +74,11 @@ public Map<String, Long> getNumberOfUpdateEventsSeen() {
return updateCount.getCounter();
}
@Override
public Map<String, Long> getNumberOfTruncateEventsSeen() {
return truncateCount.getCounter();
}
@Override
public void pause() {
isPaused = true;
@ -84,6 +93,7 @@ public void reset() {
createCount.reset();
updateCount.reset();
deleteCount.reset();
truncateCount.reset();
}
public static class ActivityCounter {

6
debezium-core/src/test/java/io/debezium/transforms/ActivityMonitoringMeterTest.java
View File

@ -26,6 +26,7 @@ public void whenNoEventIsReceivedAndNewCreatedMeterThenNoMetricsMustBeReturned()
assertThat(activityMonitoringMeter.getNumberOfCreateEventsSeen()).isEmpty();
assertThat(activityMonitoringMeter.getNumberOfUpdateEventsSeen()).isEmpty();
assertThat(activityMonitoringMeter.getNumberOfDeleteEventsSeen()).isEmpty();
assertThat(activityMonitoringMeter.getNumberOfTruncateEventsSeen()).isEmpty();
}
@ -39,6 +40,7 @@ public void whenInsertEventIsReceivedThenCorrectMetricsMustBeReturned() {
activityMonitoringMeter.onEvent(TableId.parse("db.schema.table"), null, 1L, null, Envelope.Operation.UPDATE);
activityMonitoringMeter.onEvent(TableId.parse("db.schema.table"), null, 1L, null, Envelope.Operation.DELETE);
activityMonitoringMeter.onEvent(TableId.parse("db.schema.anotherTable"), null, 1L, null, Envelope.Operation.CREATE);
activityMonitoringMeter.onEvent(TableId.parse("db.schema.anotherTable"), null, 1L, null, Envelope.Operation.TRUNCATE);
assertThat(activityMonitoringMeter.getNumberOfCreateEventsSeen())
.contains(Map.entry("db.schema.table", 2L), Map.entry("db.schema.anotherTable", 1L));
@ -46,6 +48,8 @@ public void whenInsertEventIsReceivedThenCorrectMetricsMustBeReturned() {
.contains(Map.entry("db.schema.table", 1L));
assertThat(activityMonitoringMeter.getNumberOfDeleteEventsSeen())
.contains(Map.entry("db.schema.table", 1L));
assertThat(activityMonitoringMeter.getNumberOfTruncateEventsSeen())
.contains(Map.entry("db.schema.anotherTable", 1L));
}
@ -59,11 +63,13 @@ public void whenMeterIsResetThenNoMetricsMustBeReturned() {
activityMonitoringMeter.onEvent(TableId.parse("db.schema.table"), null, 1L, null, Envelope.Operation.UPDATE);
activityMonitoringMeter.onEvent(TableId.parse("db.schema.table"), null, 1L, null, Envelope.Operation.DELETE);
activityMonitoringMeter.onEvent(TableId.parse("db.schema.anotherTable"), null, 1L, null, Envelope.Operation.CREATE);
activityMonitoringMeter.onEvent(TableId.parse("db.schema.anotherTable"), null, 1L, null, Envelope.Operation.TRUNCATE);
activityMonitoringMeter.reset();
assertThat(activityMonitoringMeter.getNumberOfCreateEventsSeen()).isEmpty();
assertThat(activityMonitoringMeter.getNumberOfUpdateEventsSeen()).isEmpty();
assertThat(activityMonitoringMeter.getNumberOfDeleteEventsSeen()).isEmpty();
assertThat(activityMonitoringMeter.getNumberOfTruncateEventsSeen()).isEmpty();
}
}