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

Remove class from DebeziumFormat

Browse Source
This commit is contained in:
Corneil du Plessis
2023-05-04 16:19:27 +02:00
parent bafa7b2133
commit 0266f5be29
2 changed files with 18 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineAutoConfiguration.java
View File

@@ -18,12 +18,15 @@ package org.springframework.cloud.fn.supplier.debezium;
import java.time.Clock;
import java.time.Duration;
import java.util.Map;
import java.util.Objects;
import java.util.function.Consumer;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.DebeziumEngine.CompletionCallback;
import io.debezium.engine.DebeziumEngine.ConnectorCallback;
import io.debezium.engine.format.SerializationFormat;
import io.debezium.engine.spi.OffsetCommitPolicy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -33,6 +36,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import static org.springframework.cloud.fn.supplier.debezium.DebeziumProperties.DebeziumFormat;
/**
* DebeziumEngine auto-configuration.
*
@@ -46,6 +51,7 @@ import org.springframework.context.annotation.Bean;
* restart.
*
* @author Christian Tzolov
* @author Corneil du Plessis
*/
@AutoConfiguration
@EnableConfigurationProperties(DebeziumProperties.class)
@@ -105,14 +111,19 @@ public class DebeziumEngineAutoConfiguration {
public ConnectorCallback connectorCallback() {
return DEFAULT_CONNECTOR_CALLBACK;
}
private static final Map<DebeziumFormat, Class<? extends SerializationFormat<byte[]>>> serialFormats = Map.of(
DebeziumFormat.JSON, io.debezium.engine.format.JsonByteArray.class,
DebeziumFormat.AVRO, io.debezium.engine.format.Avro.class,
DebeziumFormat.PROTOBUF, io.debezium.engine.format.Protobuf.class
);
@Bean
public DebeziumEngine<?> debeziumEngine(Consumer<ChangeEvent<byte[], byte[]>> changeEventConsumer,
OffsetCommitPolicy offsetCommitPolicy, CompletionCallback completionCallback,
ConnectorCallback connectorCallback, DebeziumProperties properties, Clock debeziumClock) {
Class<? extends SerializationFormat<byte[]>> format = Objects.requireNonNull(serialFormats.get(properties.getFormat()), "Cannot find format for " + properties.getProperties());
DebeziumEngine<ChangeEvent<byte[], byte[]>> debeziumEngine = DebeziumEngine
.create(properties.getFormat().serializationFormat())
.create(format)
.using(properties.getDebeziumNativeConfiguration())
.using(debeziumClock)
.using(completionCallback)

16
functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java
View File

@@ -20,8 +20,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import io.debezium.engine.format.SerializationFormat;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
@@ -34,27 +32,23 @@ public class DebeziumProperties {
/**
* JSON change event format.
*/
JSON(io.debezium.engine.format.JsonByteArray.class, "application/json"),
JSON("application/json"),
/**
* AVRO change event format.
*/
AVRO(io.debezium.engine.format.Avro.class, "application/avro"),
AVRO("application/avro"),
/**
* ProtoBuf change event format.
*/
PROTOBUF(io.debezium.engine.format.Protobuf.class, "application/x-protobuf"),;
PROTOBUF("application/x-protobuf"),;
private final Class<? extends SerializationFormat<byte[]>> serializationFormat;
private final String contentType;
DebeziumFormat(Class<? extends SerializationFormat<byte[]>> serializationFormat, String contentType) {
this.serializationFormat = serializationFormat;
DebeziumFormat(String contentType) {
this.contentType = contentType;
}
public Class<? extends SerializationFormat<byte[]>> serializationFormat() {
return serializationFormat;
}
public final String contentType() {
return contentType;