Skip to content

Commit e4259b2

Browse files
KMohnishMKMohnishM
committed
DBZ-1719 Create an ErrorHandler for Capturing destinations
Signed-off-by: KMohnishM <kmohnishm@gmail.com>
1 parent fb18f65 commit e4259b2

10 files changed

Lines changed: 588 additions & 36 deletions

File tree

‎debezium-quarkus-parent/debezium-quarkus-engine-parent/runtime/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,11 @@
4545
<artifactId>debezium-common</artifactId>
4646
<version>${project.version}</version>
4747
</dependency>
48+
<dependency>
49+
<groupId>io.debezium</groupId>
50+
<artifactId>debezium-config</artifactId>
51+
<version>${project.version}</version>
52+
</dependency>
4853
<dependency>
4954
<groupId>io.debezium</groupId>
5055
<artifactId>debezium-embedded</artifactId>

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

Lines changed: 48 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@
1212
import java.util.Optional;
1313
import java.util.stream.Collectors;
1414

15+
import jakarta.enterprise.inject.Instance;
16+
1517
import org.apache.kafka.connect.source.SourceRecord;
1618
import org.slf4j.Logger;
1719
import org.slf4j.LoggerFactory;
@@ -26,6 +28,7 @@
2628
import io.debezium.runtime.CapturingEvent;
2729
import io.debezium.runtime.CapturingEvents;
2830
import io.debezium.runtime.EngineManifest;
31+
import io.debezium.runtime.configuration.DebeziumEngineRuntimeConfiguration;
2932
import io.quarkus.debezium.engine.OperationMapper;
3033
import io.quarkus.debezium.engine.capture.CapturingEventsInvokerRegistry;
3134
import io.quarkus.debezium.engine.capture.CapturingInvoker;
@@ -42,11 +45,17 @@ public final class DefaultChangeConsumerFactory implements ChangeConsumerFactory
4245

4346
private final RoutedQuarkusChangeConsumer routedQuarkusChangeConsumer;
4447
private final Optional<CapturingTombstoneEvents> capturingTombstoneEvents;
48+
private final DebeziumEngineRuntimeConfiguration configuration;
49+
private final Instance<ErrorHandler> errorHandlers;
4550

4651
DefaultChangeConsumerFactory(CapturingEventsInvokerRegistry<CapturingEvents> filterRegistry,
4752
CapturingEventsInvokerRegistry<CapturingEvents> registry,
48-
Optional<CapturingTombstoneEvents> capturingTombstoneEvents) {
53+
Optional<CapturingTombstoneEvents> capturingTombstoneEvents,
54+
DebeziumEngineRuntimeConfiguration configuration,
55+
Instance<ErrorHandler> errorHandlers) {
4956
this.capturingTombstoneEvents = capturingTombstoneEvents;
57+
this.configuration = configuration;
58+
this.errorHandlers = errorHandlers;
5059
this.routedQuarkusChangeConsumer = new RoutedQuarkusChangeConsumer(
5160
capturingTombstoneEvents,
5261
new FilterChangeConsumer(filterRegistry),
@@ -56,9 +65,10 @@ public final class DefaultChangeConsumerFactory implements ChangeConsumerFactory
5665

5766
@Override
5867
public QuarkusChangeConsumer get(EngineManifest manifest) {
59-
return new QuarkusChangeConsumer() {
68+
QuarkusChangeConsumer baseConsumer = new QuarkusChangeConsumer() {
6069
@Override
61-
public void handleBatch(List<ChangeEvent<Object, Object>> records, DebeziumEngine.RecordCommitter<ChangeEvent<Object, Object>> committer) {
70+
public void handleBatch(List<ChangeEvent<Object, Object>> records, DebeziumEngine.RecordCommitter<ChangeEvent<Object, Object>> committer)
71+
throws InterruptedException {
6272
routedQuarkusChangeConsumer.handle(manifest, records, committer);
6373
}
6474

@@ -69,6 +79,41 @@ public boolean supportsTombstoneEvents() {
6979
.isSupported();
7080
}
7181
};
82+
83+
ErrorHandler errorHandler = ErrorHandler.resolve(manifest, configuration, errorHandlers, LoggerFactory.getLogger(DefaultChangeConsumerFactory.class));
84+
if (errorHandler != null) {
85+
return new QuarkusChangeConsumer() {
86+
@Override
87+
public void handleBatch(List<ChangeEvent<Object, Object>> records, DebeziumEngine.RecordCommitter<ChangeEvent<Object, Object>> committer)
88+
throws InterruptedException {
89+
List<BatchEvent> batchEvents = records.stream()
90+
.map(record -> new DefaultBatchEvent<>(record, committer))
91+
.collect(Collectors.toList());
92+
CapturingEvents<BatchEvent> capturingEvents = new DefaultCapturingEvents<>(batchEvents, manifest, null);
93+
try {
94+
errorHandler.handle(capturingEvents, new ErrorHandler.BatchConsumer() {
95+
@Override
96+
public void accept(CapturingEvents<BatchEvent> events) throws Exception {
97+
baseConsumer.handleBatch(records, committer);
98+
}
99+
});
100+
}
101+
catch (InterruptedException e) {
102+
throw e;
103+
}
104+
catch (Exception e) {
105+
throw new DebeziumException(e);
106+
}
107+
}
108+
109+
@Override
110+
public boolean supportsTombstoneEvents() {
111+
return baseConsumer.supportsTombstoneEvents();
112+
}
113+
};
114+
}
115+
116+
return baseConsumer;
72117
}
73118

74119
/**

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

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* Copyright Debezium Authors.
3+
*
4+
* Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
5+
*/
6+
7+
package io.quarkus.debezium.engine.capture.consumer;
8+
9+
import org.apache.kafka.connect.source.SourceRecord;
10+
import org.slf4j.Logger;
11+
import org.slf4j.LoggerFactory;
12+
13+
import io.debezium.runtime.BatchEvent;
14+
import io.debezium.runtime.CapturingEvent;
15+
import io.debezium.runtime.CapturingEvents;
16+
17+
public class DefaultRetryErrorHandler implements ErrorHandler {
18+
private static final Logger LOGGER = LoggerFactory.getLogger(DefaultRetryErrorHandler.class);
19+
20+
private final int maxRetries;
21+
private final long initialDelayMs;
22+
private final long maxDelayMs;
23+
private final double delayMultiplier;
24+
25+
public DefaultRetryErrorHandler(int maxRetries, long initialDelayMs, long maxDelayMs, double delayMultiplier) {
26+
this.maxRetries = maxRetries;
27+
this.initialDelayMs = initialDelayMs;
28+
this.maxDelayMs = maxDelayMs;
29+
this.delayMultiplier = delayMultiplier;
30+
}
31+
32+
@Override
33+
public void handle(CapturingEvents<BatchEvent> records, BatchConsumer consumer) throws Exception {
34+
int attempt = 0;
35+
long delay = initialDelayMs;
36+
37+
while (true) {
38+
try {
39+
consumer.accept(records);
40+
return;
41+
}
42+
catch (InterruptedException e) {
43+
Thread.currentThread().interrupt();
44+
throw e;
45+
}
46+
catch (Exception e) {
47+
attempt++;
48+
if (attempt > maxRetries) {
49+
LOGGER.error("Failed to process batch after {} attempts. Propagating exception.", attempt, e);
50+
throw e;
51+
}
52+
LOGGER.warn("Error processing batch (attempt {}/{}). Retrying in {} ms.", attempt, maxRetries, delay, e);
53+
Thread.sleep(delay);
54+
delay = Math.min((long) (delay * delayMultiplier), maxDelayMs);
55+
}
56+
}
57+
}
58+
59+
@Override
60+
public void handle(CapturingEvent<SourceRecord, SourceRecord> event, EventConsumer consumer) throws Exception {
61+
int attempt = 0;
62+
long delay = initialDelayMs;
63+
64+
while (true) {
65+
try {
66+
consumer.accept(event);
67+
return;
68+
}
69+
catch (InterruptedException ie) {
70+
Thread.currentThread().interrupt();
71+
throw ie;
72+
}
73+
catch (Exception e) {
74+
attempt++;
75+
if (attempt > maxRetries) {
76+
LOGGER.error("Failed to process event after {} attempts. Propagating exception.", attempt, e);
77+
throw e;
78+
}
79+
LOGGER.warn("Error processing event (attempt {}/{}). Retrying in {} ms.", attempt, maxRetries, delay, e);
80+
Thread.sleep(delay);
81+
delay = Math.min((long) (delay * delayMultiplier), maxDelayMs);
82+
}
83+
}
84+
}
85+
}

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

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
/*
2+
* Copyright Debezium Authors.
3+
*
4+
* Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
5+
*/
6+
7+
package io.quarkus.debezium.engine.capture.consumer;
8+
9+
import java.util.Optional;
10+
11+
import jakarta.enterprise.inject.Instance;
12+
import jakarta.enterprise.inject.literal.NamedLiteral;
13+
14+
import org.apache.kafka.connect.source.SourceRecord;
15+
import org.slf4j.Logger;
16+
17+
import io.debezium.runtime.BatchEvent;
18+
import io.debezium.runtime.CapturingEvent;
19+
import io.debezium.runtime.CapturingEvents;
20+
import io.debezium.runtime.EngineManifest;
21+
import io.debezium.runtime.configuration.DebeziumEngineRuntimeConfiguration;
22+
23+
public interface ErrorHandler {
24+
25+
@FunctionalInterface
26+
interface BatchConsumer {
27+
void accept(CapturingEvents<BatchEvent> events) throws Exception;
28+
}
29+
30+
@FunctionalInterface
31+
interface EventConsumer {
32+
void accept(CapturingEvent<SourceRecord, SourceRecord> event) throws Exception;
33+
}
34+
35+
default void handle(CapturingEvents<BatchEvent> records, BatchConsumer consumer) throws Exception {
36+
consumer.accept(records);
37+
}
38+
39+
default void handle(CapturingEvent<SourceRecord, SourceRecord> event, EventConsumer consumer) throws Exception {
40+
consumer.accept(event);
41+
}
42+
43+
static ErrorHandler resolve(
44+
EngineManifest manifest,
45+
DebeziumEngineRuntimeConfiguration configuration,
46+
Instance<ErrorHandler> errorHandlers,
47+
Logger logger) {
48+
if (configuration == null || configuration.capturing() == null) {
49+
return null;
50+
}
51+
52+
Optional<DebeziumEngineRuntimeConfiguration.Capturing> capturingConfig = configuration.capturing().values().stream()
53+
.filter(c -> c.engineId().isPresent() && c.engineId().get().equals(manifest.id()))
54+
.findFirst();
55+
if (capturingConfig.isEmpty()) {
56+
DebeziumEngineRuntimeConfiguration.Capturing byKey = configuration.capturing().get(manifest.id());
57+
if (byKey != null) {
58+
capturingConfig = Optional.of(byKey);
59+
}
60+
}
61+
if (capturingConfig.isEmpty() && "default".equals(manifest.id())) {
62+
DebeziumEngineRuntimeConfiguration.Capturing defaultKey = configuration.capturing().get("default");
63+
if (defaultKey != null) {
64+
capturingConfig = Optional.of(defaultKey);
65+
}
66+
}
67+
68+
ErrorHandler errorHandler = null;
69+
70+
if (capturingConfig.isPresent()) {
71+
DebeziumEngineRuntimeConfiguration.Capturing cap = capturingConfig.get();
72+
if (cap.errorHandler().isPresent()) {
73+
String handlerName = cap.errorHandler().get();
74+
Instance<ErrorHandler> namedInstance = errorHandlers.select(NamedLiteral.of(handlerName));
75+
if (namedInstance.isResolvable()) {
76+
errorHandler = namedInstance.get();
77+
}
78+
else {
79+
try {
80+
Class<?> clazz = Class.forName(handlerName, true, Thread.currentThread().getContextClassLoader());
81+
errorHandler = (ErrorHandler) clazz.getDeclaredConstructor().newInstance();
82+
}
83+
catch (Exception e) {
84+
throw new RuntimeException("Could not find or instantiate ErrorHandler: " + handlerName, e);
85+
}
86+
}
87+
}
88+
else {
89+
if (!errorHandlers.isUnsatisfied()) {
90+
if (errorHandlers.isResolvable()) {
91+
errorHandler = errorHandlers.get();
92+
}
93+
else {
94+
logger.warn("Multiple ErrorHandler CDI beans found but none specifically configured for engine. Using none.");
95+
}
96+
}
97+
98+
if (errorHandler == null) {
99+
int maxRetries = cap.maxRetries().orElse(0);
100+
if (maxRetries > 0) {
101+
long initialDelayMs = cap.initialDelayMs().orElse(1000L);
102+
long maxDelayMs = cap.maxDelayMs().orElse(10000L);
103+
double delayMultiplier = cap.delayMultiplier().orElse(2.0);
104+
errorHandler = new DefaultRetryErrorHandler(maxRetries, initialDelayMs, maxDelayMs, delayMultiplier);
105+
}
106+
}
107+
}
108+
}
109+
else {
110+
if (!errorHandlers.isUnsatisfied()) {
111+
if (errorHandlers.isResolvable()) {
112+
errorHandler = errorHandlers.get();
113+
}
114+
else {
115+
logger.warn("Multiple ErrorHandler CDI beans found but none specifically configured for engine. Using none.");
116+
}
117+
}
118+
}
119+
120+
return errorHandler;
121+
}
122+
}

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

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import jakarta.inject.Named;
1515

1616
import io.debezium.runtime.CapturingEvents;
17+
import io.debezium.runtime.configuration.DebeziumEngineRuntimeConfiguration;
1718
import io.quarkus.debezium.engine.capture.CapturingEventsInvokerRegistry;
1819
import io.quarkus.debezium.engine.capture.CapturingTombstoneEvents;
1920

@@ -35,12 +36,12 @@ public GeneralChangeConsumerProducer(@Named("capturingFilterEventsInvokerRegistr
3536

3637
@Produces
3738
@Dependent
38-
public ChangeConsumerFactory produce() {
39+
public ChangeConsumerFactory produce(DebeziumEngineRuntimeConfiguration configuration, Instance<ErrorHandler> errorHandlers) {
3940
if (registry == null) {
4041
return null;
4142
}
4243

43-
return new DefaultChangeConsumerFactory(filterRegistry, registry, capturingTombstoneEvents);
44+
return new DefaultChangeConsumerFactory(filterRegistry, registry, capturingTombstoneEvents, configuration, errorHandlers);
4445
}
4546

4647
}

0 commit comments

Comments
 (0)