SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Try to mitigate cdc overflow error

Browse Source
This commit is contained in:
Christian Tzolov
2021-06-09 15:19:19 +02:00
parent 5327a5b5e3
commit ed77484f72
1 changed files with 2 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
functions/supplier/cdc-debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/cdc/CdcSupplierConfiguration.java
View File

@@ -31,6 +31,7 @@ import org.apache.kafka.connect.source.SourceRecord;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Sinks;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -114,7 +115,7 @@ public class CdcSupplierConfiguration implements BeanClassLoaderAware {
Function<SourceRecord, SourceRecord> recordFlattening,
ObjectMapper mapper, CdcSupplierProperties cdcStreamingEngineProperties) {
FluxSink<Message<?>> sink = emitterProcessor.sink();
FluxSink<Message<?>> sink = emitterProcessor.sink(FluxSink.OverflowStrategy.BUFFER);
Consumer<SourceRecord> messageConsumer = sourceRecord -> {
// When cdc.flattening.deleteHandlingMode=none and cdc.flattening.dropTombstones=false