Skip to content

Commit e3cf732

Browse files
vsantonastasokmos
vsantonastaso
authored and
kmos
committed
debezium/dbz#1081 add capturing filter strategy for single event processing
Signed-off-by: vsantonastaso <vsantonastaso.dev@gmail.com>
1 parent 5080411 commit e3cf732

27 files changed

Lines changed: 1247 additions & 59 deletions

‎debezium-quarkus-parent/debezium-quarkus-engine-parent/deployment/src/main/java/io/quarkus/debezium/deployment/engine/CapturingEventGenerator.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,10 +41,11 @@ public boolean isCompatible(Type type) {
4141
}
4242

4343
/**
44-
* it generates concrete classes based on the {@link io.quarkus.debezium.engine.capture.CapturingEventInvoker} interface using gizmo:
44+
* it generates concrete classes based on the {@link CapturingEventInvoker} interface using gizmo:
4545
* <p>
4646
* public class GeneratedCapturingInvoker {
4747
* private final Object beanInstance;
48+
* private final Object filter;
4849
* <p>
4950
* void capture(CapturingEvent<SourceRecord> event) {
5051
* beanInstance.method(event);
@@ -54,10 +55,11 @@ public boolean isCompatible(Type type) {
5455
*
5556
* @param methodInfo
5657
* @param beanInfo
58+
* @param filter
5759
* @return
5860
*/
5961
@Override
60-
public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo) {
62+
public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo, BeanInfo filter) {
6163
String name = generateClassName(beanInfo, methodInfo);
6264

6365
try (ClassCreator invoker = ClassCreator.builder()
@@ -84,6 +86,7 @@ public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo)
8486
}
8587

8688
createEngineMethod(methodInfo, invoker);
89+
handleFilterMethod(filter, invoker, beanInstanceField);
8790

8891
return new GeneratedClassMetaData(UUID.randomUUID(), name.replace('/', '.'), beanInfo, CapturingEventInvoker.class);
8992
}

‎debezium-quarkus-parent/debezium-quarkus-engine-parent/deployment/src/main/java/io/quarkus/debezium/deployment/engine/CapturingEventsGenerator.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,23 +40,29 @@ public boolean isCompatible(Type type) {
4040
}
4141

4242
/**
43-
* it generates concrete classes based on the {@link io.quarkus.debezium.engine.capture.CapturingEventInvoker} interface using gizmo:
43+
* it generates concrete classes based on the {@link CapturingEventsInvoker} interface using gizmo:
4444
* <p>
4545
* public class GeneratedCapturingInvoker {
4646
* private final Object beanInstance;
47+
* private final Object filter;
4748
* <p>
4849
* void capture(CapturingEvents<BatchEvent> event) {
4950
* beanInstance.method(event);
5051
* }
5152
* <p>
53+
* boolean shouldCapture(CapturingEvent event) {
54+
* return filter.shouldCapture(event);
55+
* }
56+
* <p>
5257
* }
5358
*
5459
* @param methodInfo
5560
* @param beanInfo
61+
* @param filter
5662
* @return
5763
*/
5864
@Override
59-
public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo) {
65+
public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo, BeanInfo filter) {
6066
String name = generateClassName(beanInfo, methodInfo);
6167

6268
try (ClassCreator invoker = ClassCreator.builder()
@@ -83,6 +89,7 @@ public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo)
8389
}
8490

8591
createEngineMethod(methodInfo, invoker);
92+
handleFilterMethod(filter, invoker, beanInstanceField);
8693

8794
return new GeneratedClassMetaData(UUID.randomUUID(), name.replace('/', '.'), beanInfo, CapturingEventsInvoker.class);
8895
}

‎debezium-quarkus-parent/debezium-quarkus-engine-parent/deployment/src/main/java/io/quarkus/debezium/deployment/engine/CapturingInvokerGenerator.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,5 +24,5 @@ default boolean isCompatible(Type type) {
2424
return false;
2525
}
2626

27-
GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo);
27+
GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo, BeanInfo filter);
2828
}

‎debezium-quarkus-parent/debezium-quarkus-engine-parent/deployment/src/main/java/io/quarkus/debezium/deployment/engine/CapturingObjectInvokerGenerator.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,23 +38,29 @@ public boolean isCompatible(Type type) {
3838
}
3939

4040
/**
41-
* it generates concrete classes based on the {@link io.quarkus.debezium.engine.capture.CapturingObjectInvoker} interface using gizmo:
41+
* it generates concrete classes based on the {@link CapturingObjectInvoker} interface using gizmo:
4242
* <p>
4343
* public class GeneratedCapturingInvoker {
4444
* private final Object beanInstance;
45+
* private final Object filter;
4546
* <p>
4647
* void capture(Object event) {
4748
* beanInstance.method(event);
4849
* }
4950
* <p>
51+
* boolean shouldCapture(CapturingEvent event) {
52+
* return filter.shouldCapture(event);
53+
* }
54+
* <p>
5055
* }
5156
*
5257
* @param methodInfo
5358
* @param beanInfo
59+
* @param filter
5460
* @return
5561
*/
5662
@Override
57-
public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo) {
63+
public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo, BeanInfo filter) {
5864
String name = generateClassName(beanInfo, methodInfo);
5965

6066
try (ClassCreator invoker = ClassCreator.builder()
@@ -84,6 +90,7 @@ public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo)
8490
}
8591

8692
createEngineMethod(methodInfo, invoker);
93+
handleFilterMethod(filter, invoker, beanInstanceField);
8794

8895
return new GeneratedClassMetaData(UUID.randomUUID(), name.replace('/', '.'), beanInfo, CapturingObjectInvoker.class);
8996
}

‎debezium-quarkus-parent/debezium-quarkus-engine-parent/deployment/src/main/java/io/quarkus/debezium/deployment/engine/EngineProcessor.java‎

Lines changed: 39 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
import io.debezium.pipeline.signal.channels.process.InProcessSignalChannel;
5555
import io.debezium.pipeline.txmetadata.DefaultTransactionMetadataFactory;
5656
import io.debezium.processors.spi.PostProcessor;
57+
import io.debezium.runtime.CapturingFilterStrategy;
5758
import io.debezium.runtime.DebeziumConnectorRegistry;
5859
import io.debezium.runtime.FieldFilterStrategy;
5960
import io.debezium.runtime.configuration.DebeziumEngineBuildTimeConfiguration;
@@ -94,6 +95,7 @@
9495
import io.quarkus.debezium.engine.DefaultStateHandler;
9596
import io.quarkus.debezium.engine.capture.CapturingEventInvokerRegistryProducer;
9697
import io.quarkus.debezium.engine.capture.CapturingEventsInvokerRegistryProducer;
98+
import io.quarkus.debezium.engine.capture.CapturingFilterInvokerRegistryProducer;
9799
import io.quarkus.debezium.engine.capture.CapturingInvoker;
98100
import io.quarkus.debezium.engine.capture.CapturingObjectInvokerRegistryProducer;
99101
import io.quarkus.debezium.engine.capture.DynamicCapturingInvokerSupplier;
@@ -172,6 +174,7 @@ void engine(BuildProducer<AdditionalBeanBuildItem> additionalBeanProducer,
172174
.builder()
173175
.addBeanClasses(
174176
CapturingEventInvokerRegistryProducer.class,
177+
CapturingFilterInvokerRegistryProducer.class,
175178
CapturingObjectInvokerRegistryProducer.class,
176179
CapturingEventsInvokerRegistryProducer.class)
177180
.setDefaultScope(DotNames.APPLICATION_SCOPED)
@@ -456,12 +459,32 @@ public void generateInvokers(List<DebeziumMediatorBuildItem> mediatorBuildItems,
456459
.filter(item -> item.getDotName().equals(DebeziumDotNames.FIELD_FILTER_STRATEGY))
457460
.collect(Collectors.toMap(item -> item.getBean().getProviderType(), item -> item));
458461

462+
Map<Type, DebeziumMediatorBuildItem> capturingFilterByClassName = mediatorBuildItems.stream()
463+
.filter(item -> item.getDotName().equals(DebeziumDotNames.CAPTURING_FILTER_STRATEGY))
464+
.collect(Collectors.toMap(item -> item.getBean().getProviderType(), item -> item));
465+
459466
mediatorBuildItems.forEach(item -> {
460467
if (item.getDotName().equals(DebeziumDotNames.CAPTURING)) {
468+
BeanInfo filter = Optional.ofNullable(item.getMethodInfo().annotation(DebeziumDotNames.CAPTURING).value("filter"))
469+
.map(AnnotationValue::asClass)
470+
.map(capturingFilterByClassName::get)
471+
.map(DebeziumMediatorBuildItem::getBean)
472+
.orElse(null);
473+
474+
if (filter != null) {
475+
var annotation = item.getMethodInfo().annotation(DebeziumDotNames.CAPTURING);
476+
if (annotation.value("destination") != null || annotation.value("engine") != null) {
477+
throw new IllegalArgumentException(
478+
"@Capturing on method " + item.getMethodInfo().declaringClass().name() + "."
479+
+ item.getMethodInfo().name()
480+
+ " cannot set 'destination' or 'engine' when 'filter' is specified");
481+
}
482+
}
483+
461484
GeneratedClassMetaData metadata = capturingInvokerGenerator.generate(item.getMethodInfo(),
462-
item.getBean());
485+
item.getBean(), filter);
463486
debeziumGeneratedInvokerBuildItemBuildProducer.produce(new DebeziumGeneratedInvokerBuildItem(metadata.generatedClassName(),
464-
metadata.mediator(), metadata.getShortIdentifier(), metadata.clazz()));
487+
metadata.mediator(), metadata.getShortIdentifier(), metadata.clazz(), filter));
465488
reflectiveClassBuildItemBuildProducer.produce(ReflectiveClassBuildItem.builder(metadata.generatedClassName()).build());
466489
}
467490

@@ -510,6 +533,7 @@ public void injectInvokers(
510533
.unremovable()
511534
.supplier(dynamicCapturingInvokerSupplier.createInvoker(
512535
recorderContext.classProxy(item.getMediator().getImplClazz().name().toString()),
536+
item.getFilter() != null ? recorderContext.classProxy(item.getFilter().getImplClazz().name().toString()) : null,
513537
(Class<? extends CapturingInvoker<?>>) recorderContext.classProxy(item.getGeneratedClassName())))
514538
.named(DynamicCapturingInvokerSupplier.BASE_NAME + item.getMediator().getImplClazz().name() + item.getId())
515539
.done()));
@@ -581,6 +605,15 @@ public void extractMediators(BuildProducer<DebeziumMediatorBuildItem> mediatorBu
581605
.stream()
582606
.map(info -> new DebeziumMediatorBuildItem(info, null, DebeziumDotNames.FIELD_FILTER_STRATEGY))
583607
.forEach(mediatorBuildItemBuildProducer::produce);
608+
609+
beanDiscoveryFinished
610+
.beanStream()
611+
.classBeans()
612+
.filter(info -> info.getTypes().stream()
613+
.anyMatch(type -> type.name().equals(DebeziumDotNames.CAPTURING_FILTER_STRATEGY)))
614+
.stream()
615+
.map(info -> new DebeziumMediatorBuildItem(info, null, DebeziumDotNames.CAPTURING_FILTER_STRATEGY))
616+
.forEach(mediatorBuildItemBuildProducer::produce);
584617
}
585618

586619
@BuildStep
@@ -592,8 +625,10 @@ public List<UnremovableBeanBuildItem> removalExclusion() {
592625
}
593626

594627
@BuildStep
595-
public UnremovableBeanBuildItem avoidRemovalIfNotReferenced() {
596-
return UnremovableBeanBuildItem.beanTypes(FieldFilterStrategy.class);
628+
public List<UnremovableBeanBuildItem> avoidRemovalIfNotReferenced() {
629+
return List.of(
630+
UnremovableBeanBuildItem.beanTypes(FieldFilterStrategy.class),
631+
UnremovableBeanBuildItem.beanTypes(CapturingFilterStrategy.class));
597632
}
598633

599634
private Optional<String> extractSourceConnector(Path path) {

‎debezium-quarkus-parent/debezium-quarkus-engine-parent/deployment/src/main/java/io/quarkus/debezium/deployment/engine/GizmoBasedCapturingInvokerGenerator.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@
1212
import org.jboss.jandex.MethodInfo;
1313

1414
import io.debezium.runtime.Capturing;
15+
import io.debezium.runtime.CapturingEvent;
16+
import io.quarkus.arc.processor.BeanInfo;
1517
import io.quarkus.debezium.deployment.dotnames.DebeziumDotNames;
1618
import io.quarkus.gizmo.ClassCreator;
1719
import io.quarkus.gizmo.FieldDescriptor;
@@ -75,4 +77,62 @@ default void createEngineMethod(MethodInfo methodInfo, ClassCreator invoker) {
7577
() -> engine.returnValue(engine.load(Capturing.DEFAULT)));
7678
}
7779
}
80+
81+
default void handleFilterMethod(BeanInfo filter, ClassCreator invoker, FieldDescriptor beanInstanceField) {
82+
boolean hasFilter = filter != null;
83+
createHasFilterMethod(invoker, hasFilter);
84+
if (hasFilter) {
85+
FieldDescriptor filterInstanceField = createFilterField(invoker, beanInstanceField, filter);
86+
createShouldCaptureMethod(invoker, filter, filterInstanceField);
87+
}
88+
}
89+
90+
private FieldDescriptor createFilterField(ClassCreator invoker, FieldDescriptor beanInstanceField, BeanInfo filter) {
91+
FieldDescriptor filterInstanceField = invoker.getFieldCreator("filter", filter
92+
.getImplClazz()
93+
.name()
94+
.toString())
95+
.setModifiers(Modifier.PRIVATE)
96+
.getFieldDescriptor();
97+
98+
try (MethodCreator constructor = invoker.getMethodCreator("<init>", void.class, Object.class, Object.class)) {
99+
constructor.setModifiers(Modifier.PUBLIC);
100+
constructor.invokeSpecialMethod(MethodDescriptor.ofConstructor(Object.class), constructor.getThis());
101+
ResultHandle constructorThis = constructor.getThis();
102+
103+
ResultHandle beanInstance = constructor.getMethodParam(0);
104+
constructor.writeInstanceField(beanInstanceField, constructorThis, beanInstance);
105+
106+
ResultHandle filterInstance = constructor.getMethodParam(1);
107+
constructor.writeInstanceField(filterInstanceField, constructorThis, filterInstance);
108+
109+
constructor.returnValue(null);
110+
}
111+
112+
return filterInstanceField;
113+
}
114+
115+
private void createHasFilterMethod(ClassCreator invoker, boolean hasFilter) {
116+
try (MethodCreator method = invoker.getMethodCreator("hasFilter", boolean.class)) {
117+
method.returnValue(method.load(hasFilter));
118+
}
119+
}
120+
121+
private void createShouldCaptureMethod(ClassCreator invoker, BeanInfo filter, FieldDescriptor filterInstanceField) {
122+
try (MethodCreator shouldCapture = invoker.getMethodCreator("shouldCapture", boolean.class, CapturingEvent.class)) {
123+
ResultHandle filterThis = shouldCapture.getThis();
124+
ResultHandle delegate = shouldCapture.readInstanceField(filterInstanceField, filterThis);
125+
ResultHandle event = shouldCapture.getMethodParam(0);
126+
127+
MethodDescriptor methodDescriptor = MethodDescriptor.ofMethod(
128+
filter.getImplClazz().toString(),
129+
"shouldCapture",
130+
boolean.class,
131+
CapturingEvent.class);
132+
133+
ResultHandle result = shouldCapture.invokeVirtualMethod(methodDescriptor, delegate, event);
134+
shouldCapture.returnValue(result);
135+
}
136+
}
137+
78138
}

‎debezium-quarkus-parent/debezium-quarkus-engine-parent/deployment/src/main/java/io/quarkus/debezium/deployment/engine/MultipleCapturingInvokerGenerators.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,13 +24,13 @@ public MultipleCapturingInvokerGenerators(CapturingInvokerGenerator defaultGener
2424
}
2525

2626
@Override
27-
public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo) {
27+
public GeneratedClassMetaData generate(MethodInfo methodInfo, BeanInfo beanInfo, BeanInfo filter) {
2828
return generators
2929
.stream()
3030
.filter(generator -> generator.isCompatible(methodInfo.parameters().getFirst().type()))
3131
.findFirst()
32-
.map(generator -> generator.generate(methodInfo, beanInfo))
33-
.orElseGet(() -> defaultGenerator.generate(methodInfo, beanInfo));
32+
.map(generator -> generator.generate(methodInfo, beanInfo, filter))
33+
.orElseGet(() -> defaultGenerator.generate(methodInfo, beanInfo, filter));
3434
}
3535

3636
}

‎debezium-quarkus-parent/debezium-quarkus-engine-parent/runtime/src/main/java/io/quarkus/debezium/engine/capture/CapturingEventInvokerRegistryProducer.java‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,35 +6,46 @@
66

77
package io.quarkus.debezium.engine.capture;
88

9+
import static io.quarkus.debezium.engine.capture.CapturingInvoker.CapturingInvokerKey;
10+
import static io.quarkus.debezium.engine.capture.CapturingInvoker.getAllDestinations;
11+
import static io.quarkus.debezium.engine.capture.CapturingInvoker.getKey;
12+
913
import java.util.Map;
14+
import java.util.Optional;
1015
import java.util.function.Function;
16+
import java.util.function.Predicate;
1117
import java.util.stream.Collectors;
1218

1319
import jakarta.enterprise.inject.Instance;
1420
import jakarta.enterprise.inject.Produces;
1521
import jakarta.inject.Inject;
22+
import jakarta.inject.Named;
1623
import jakarta.inject.Singleton;
1724

1825
import io.debezium.runtime.CapturingEvent;
1926

2027
public class CapturingEventInvokerRegistryProducer {
28+
public static final Predicate<CapturingEventInvoker> NO_CAPTURING_FILTER_PREDICATE = inv -> !inv.hasFilter();
2129
private final CapturingInvokerValidator<CapturingEventInvoker> validator = new CapturingInvokerValidator<>();
2230

2331
@Inject
2432
Instance<CapturingEventInvoker> invokers;
2533

2634
@Produces
2735
@Singleton
36+
@Named("capturingEventInvokerRegistry")
2837
public CapturingInvokerRegistry<CapturingEvent<Object, Object>> produce() {
2938
validator.validate(this.invokers
3039
.stream()
40+
.filter(NO_CAPTURING_FILTER_PREDICATE)
3141
.toList());
3242

33-
Map<String, CapturingEventInvoker> invokers = this.invokers
43+
Map<CapturingInvokerKey, CapturingEventInvoker> invokers = this.invokers
3444
.stream()
45+
.filter(NO_CAPTURING_FILTER_PREDICATE)
3546
.collect(Collectors.toMap(CapturingInvoker::generateKey, Function.identity()));
3647

37-
return event -> invokers.getOrDefault(CapturingInvoker.getKey(event), invokers.get(CapturingInvoker.getAllDestinations(event)));
48+
return event -> Optional.ofNullable(invokers.getOrDefault(getKey(event), invokers.get(getAllDestinations(event))));
3849
}
3950

4051
}

‎debezium-quarkus-parent/debezium-quarkus-engine-parent/runtime/src/main/java/io/quarkus/debezium/engine/capture/CapturingEventsInvokerRegistryProducer.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import jakarta.inject.Inject;
1717

1818
import io.debezium.runtime.CapturingEvents;
19+
import io.quarkus.debezium.engine.capture.CapturingInvoker.CapturingInvokerKey;
1920

2021
public class CapturingEventsInvokerRegistryProducer {
2122

@@ -25,15 +26,15 @@ public class CapturingEventsInvokerRegistryProducer {
2526
@Produces
2627
@Dependent
2728
public CapturingEventsInvokerRegistry<CapturingEvents<Object>> produce() {
28-
Map<String, CapturingEventsInvoker> handlers = this.invokers
29+
Map<CapturingInvokerKey, CapturingEventsInvoker> handlers = this.invokers
2930
.stream()
3031
.collect(Collectors.toMap(CapturingInvoker::generateKey, Function.identity()));
3132

3233
if (handlers.isEmpty()) {
3334
return null;
3435
}
3536

36-
return events -> handlers.getOrDefault(events.engine() + "_" + events.destination(),
37+
return events -> handlers.getOrDefault(CapturingInvoker.getKey(events),
3738
handlers.get(CapturingInvoker.getAllDestinations(events)));
3839
}
3940
}

0 commit comments

Comments
 (0)