Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

GH-5: Use DebeziumMessageProducer from Spring Integration #33

Merged
merged 2 commits into from
Feb 26, 2024
Merged

GH-5: Use DebeziumMessageProducer from Spring Integration #33

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d7f4862
GH-5: Use `DebeziumMessageProducer` from Spring Integration
artembilan Feb 23, 2024
b8a735e
Fix typo in the `DebeziumSupplierIntegrationTests`
artembilan Feb 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions supplier/spring-debezium-supplier/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ repositories {

dependencies {
api project(':spring-debezium-autoconfigure')
api 'org.springframework.integration:spring-integration-debezium'
api 'io.debezium:debezium-connector-mysql'
api 'io.debezium:debezium-connector-mongodb'
api 'io.debezium:debezium-connector-postgres'
Expand Down
156 changes: 22 additions & 134 deletions ...org/springframework/cloud/fn/supplier/debezium/DebeziumReactiveConsumerConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,35 +16,25 @@

package org.springframework.cloud.fn.supplier.debezium;

import java.lang.reflect.Field;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
import java.util.function.Supplier;

import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.DebeziumEngine.Builder;
import io.debezium.engine.Header;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.common.config.ComponentCustomizer;
import org.springframework.cloud.fn.common.debezium.DebeziumEngineBuilderAutoConfiguration;
import org.springframework.cloud.fn.common.debezium.DebeziumProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.debezium.dsl.Debezium;
import org.springframework.integration.debezium.dsl.DebeziumMessageProducerSpec;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.ClassUtils;
import org.springframework.util.MimeTypeUtils;

/**
* The Debezium supplier auto-configuration.
Expand All @@ -55,131 +45,29 @@
@AutoConfiguration(after = DebeziumEngineBuilderAutoConfiguration.class)
@EnableConfigurationProperties(DebeziumSupplierProperties.class)
@ConditionalOnBean(DebeziumEngine.Builder.class)
public class DebeziumReactiveConsumerConfiguration implements BeanClassLoaderAware {

private static final Log LOGGER = LogFactory.getLog(DebeziumReactiveConsumerConfiguration.class);

/**
* ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL.
*/
public static final String ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL = "org.springframework.kafka.support.KafkaNull";

private Object kafkaNull = null;

@Override
public void setBeanClassLoader(ClassLoader classLoader) {
try {
Class<?> clazz = ClassUtils.forName(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, classLoader);
Field field = clazz.getDeclaredField("INSTANCE");
this.kafkaNull = field.get(null);
}
catch (ClassNotFoundException | NoSuchFieldException | IllegalAccessException ex) {
}
}

/**
* Reactive Streams, single subscriber, sink used to push down the change event
* signals received from the Debezium Engine.
*/
private final Sinks.Many<Message<?>> eventSink = Sinks.many().unicast().onBackpressureError();

/**
* Debezium Engine is designed to be submitted to an {@link ExecutorService} for
* execution by a single thread, and a running connector can be stopped either by
* calling {@code stop()} from another thread or by interrupting the running thread
* (e.g., as is the case with {@link ExecutorService#shutdownNow()}).
*/
private final ExecutorService debeziumExecutor = Executors.newSingleThreadExecutor();
public class DebeziumReactiveConsumerConfiguration {

@Bean
public DebeziumEngine<ChangeEvent<byte[], byte[]>> debeziumEngine(
Consumer<ChangeEvent<byte[], byte[]>> changeEventConsumer,
Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder) {

return debeziumEngineBuilder.notifying(changeEventConsumer).build();
public Supplier<Flux<Message<?>>> debeziumSupplier(Publisher<Message<?>> debeziumPublisher) {
return () -> Flux.from(debeziumPublisher);
}

@Bean
public Supplier<Flux<Message<?>>> debeziumSupplier(DebeziumEngine<ChangeEvent<byte[], byte[]>> debeziumEngine) {

return () -> this.eventSink.asFlux()
.doOnRequest((r) -> this.debeziumExecutor.execute(debeziumEngine))
.doOnTerminate(this.debeziumExecutor::shutdownNow);
}

@Bean
@ConditionalOnMissingBean
public Consumer<ChangeEvent<byte[], byte[]>> changeEventConsumer(DebeziumProperties engineProperties,
DebeziumSupplierProperties supplierProperties) {

return new ChangeEventConsumer<>(engineProperties.getPayloadFormat().contentType(),
supplierProperties.isCopyHeaders(), this.eventSink);
}

/**
* Format-agnostic change event consumer.
*/
private final class ChangeEventConsumer<T> implements Consumer<ChangeEvent<T, T>> {

private final String contentType;

private final boolean copyHeaders;

private final Sinks.Many<Message<?>> eventSink;

private ChangeEventConsumer(String contentType, boolean copyHeaders, Sinks.Many<Message<?>> eventSink) {
this.contentType = contentType;
this.copyHeaders = copyHeaders;
this.eventSink = eventSink;
}

@Override
public void accept(ChangeEvent<T, T> changeEvent) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("[Debezium Event]: " + changeEvent.key());
}

Object key = changeEvent.key();
Object payload = changeEvent.value();
String destination = changeEvent.destination();

// When the tombstone event is enabled, Debezium serializes the payload to
// null (e.g. empty payload)
// while the metadata information is carried through the headers
// (debezium_key).
// Note: Event for none flattened responses, when the
// debezium.properties.tombstones.on.delete=true
// (default), tombstones are generate by Debezium and handled by the code
// below.
if (payload == null) {
payload = DebeziumReactiveConsumerConfiguration.this.kafkaNull;
}

// If payload is still null ignore the message.
if (payload == null) {
LOGGER.info("Dropped null payload message");
return;
}

MessageBuilder<?> messageBuilder = MessageBuilder.withPayload(payload)
.setHeader("debezium_key", key)
.setHeader("debezium_destination", destination)
.setHeader(MessageHeaders.CONTENT_TYPE,
(payload.equals(DebeziumReactiveConsumerConfiguration.this.kafkaNull))
? MimeTypeUtils.TEXT_PLAIN_VALUE : this.contentType);

if (this.copyHeaders) {
List<Header<T>> headers = changeEvent.headers();
if (headers != null && !headers.isEmpty()) {
for (Header<T> header : headers) {
messageBuilder.setHeader(header.getKey(), header.getValue());
}
}
}

this.eventSink.tryEmitNext(messageBuilder.build());
public Publisher<Message<byte[]>> debeziumPublisher(Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder,
DebeziumSupplierProperties supplierProperties,
@Nullable ComponentCustomizer<DebeziumMessageProducerSpec> debeziumMessageProducerSpecComponentCustomizer) {

DebeziumMessageProducerSpec debeziumMessageProducerSpec = Debezium.inboundChannelAdapter(debeziumEngineBuilder)
.enableEmptyPayload(supplierProperties.isEnableEmptyPayload())
.headerNames(supplierProperties.getHeaderNamesToMap())
// TODO until Spring Integration 6.3.0-M2
.autoStartup(false);

if (debeziumMessageProducerSpecComponentCustomizer != null) {
debeziumMessageProducerSpecComponentCustomizer.customize(debeziumMessageProducerSpec);
}

return IntegrationFlow.from(debeziumMessageProducerSpec).toReactivePublisher(true);
}

}
26 changes: 20 additions & 6 deletions .../main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumSupplierProperties.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,21 +22,35 @@
* Debezium supplier configuration properties.
*
* @author Christian Tzolov
* @author Artem Bilan
*/
@ConfigurationProperties("debezium.supplier")
public class DebeziumSupplierProperties {

/**
* Copy Change Event headers into Message headers.
* Enable support for tombstone (aka delete) messages.
*/
private boolean copyHeaders = true;
private boolean enableEmptyPayload = true;

public boolean isCopyHeaders() {
return this.copyHeaders;
/**
* Patterns for {@code ChangeEvent.headers()} to map.
*/
private String[] headerNamesToMap = { "*" };

public boolean isEnableEmptyPayload() {
return this.enableEmptyPayload;
}

public void setEnableEmptyPayload(boolean enableEmptyPayload) {
this.enableEmptyPayload = enableEmptyPayload;
}

public String[] getHeaderNamesToMap() {
return this.headerNamesToMap;
}

public void setCopyHeaders(boolean copyHeaders) {
this.copyHeaders = copyHeaders;
public void setHeaderNamesToMap(String[] headerNamesToMap) {
this.headerNamesToMap = headerNamesToMap;
}

}
8 changes: 5 additions & 3 deletions ...rk/cloud/fn/supplier/debezium/it/supplier/DebeziumReactiveConsumerConfigurationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,8 @@

package org.springframework.cloud.fn.supplier.debezium.it.supplier;

import io.debezium.engine.DebeziumEngine;
import java.util.function.Supplier;

import org.junit.jupiter.api.Test;

import org.springframework.boot.autoconfigure.AutoConfigurations;
Expand All @@ -30,6 +31,7 @@
* Tests for {@link DebeziumReactiveConsumerConfiguration}.
*
* @author Christian Tzolov
* @author Artem Bilan
*/
public class DebeziumReactiveConsumerConfigurationTests {

Expand All @@ -41,13 +43,13 @@ public class DebeziumReactiveConsumerConfigurationTests {

@Test
void noConnectorNoProperty() {
this.contextRunner.run((context) -> assertThat(context).doesNotHaveBean(DebeziumEngine.class));
this.contextRunner.run((context) -> assertThat(context).doesNotHaveBean(Supplier.class));
}

@Test
void withConnectorWithProperty() {
this.contextRunner.withPropertyValues("debezium.properties.connector.class=Dummy")
.run((context) -> assertThat(context).hasSingleBean(DebeziumEngine.class));
.run((context) -> assertThat(context).hasSingleBean(Supplier.class));
}

}
46 changes: 23 additions & 23 deletions ...ingframework/cloud/fn/supplier/debezium/it/supplier/DebeziumSupplierIntegrationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

package org.springframework.cloud.fn.supplier.debezium.it.supplier;

import java.time.Duration;
import java.util.function.Supplier;

import org.junit.jupiter.api.Test;
Expand All @@ -41,16 +42,16 @@
properties = { "spring.cloud.function.definition=debeziumSupplier",

// https://debezium.io/documentation/reference/transformations/event-flattening.html
"debezium.properties.debezium.embedded.shutdown.pause.before.interrupt.ms=500",
"debezium.properties.transforms=unwrap",
"debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState",
"debezium.properties.transforms.unwrap.drop.tombstones=true",
"debezium.properties.transforms.unwrap.delete.handling.mode=rewrite",
"debezium.properties.transforms.unwrap.delete.tombstone.handling.mode=rewrite",
"debezium.properties.transforms.unwrap.add.fields=name,db,op,table",

"debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory",
"debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore",

// Drop schema from the message payload.
// Drop schema from the payload payload.
artembilan marked this conversation as resolved.
Show resolved Hide resolved
"debezium.properties.key.converter.schemas.enable=false",
"debezium.properties.value.converter.schemas.enable=false",

Expand Down Expand Up @@ -101,52 +102,51 @@ void testDebeziumSupplier() {
jdbcTemplate.update(
"INSERT INTO `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', '[email protected]')");

Flux<Message<?>> messageFlux = this.debeziumSupplier.get();
Flux<String> payloadFlux = this.debeziumSupplier.get()
.map(Message::getPayload)
.cast(byte[].class)
.map(String::new);

// Message size should correspond to the number of insert statements in:
// payload size should correspond to the number of insert statements in:
// https://github.com/debezium/container-images/blob/main/examples/mysql/2.3/inventory.sql
// filtered by Customers and Addresses table.
StepVerifier.create(messageFlux)
StepVerifier.create(payloadFlux)
.expectNextCount(16) // Skip the DDL transaction logs.

// Customers table
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":1001,\"first_name\":\"Sally\",\"last_name\":\"Thomas\",\"email\":\"[email protected]\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"customers\"}"))
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":1002,\"first_name\":\"George\",\"last_name\":\"Bailey\",\"email\":\"[email protected]\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"customers\"}"))
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":1003,\"first_name\":\"Edward\",\"last_name\":\"Walker\",\"email\":\"[email protected]\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"customers\"}"))
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":1004,\"first_name\":\"Anne\",\"last_name\":\"Kretchmar\",\"email\":\"[email protected]\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"customers\"}"))

// NEW Customer Insert
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":1005,\"first_name\":\"Test666\",\"last_name\":\"Test666\",\"email\":\"[email protected]\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"customers\"}"))

// Addresses table
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":10,\"customer_id\":1001,\"street\":\"3183 Moore Avenue\",\"city\":\"Euless\",\"state\":\"Texas\",\"zip\":\"76036\",\"type\":\"SHIPPING\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"addresses\"}"))
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":11,\"customer_id\":1001,\"street\":\"2389 Hidden Valley Road\",\"city\":\"Harrisburg\",\"state\":\"Pennsylvania\",\"zip\":\"17116\",\"type\":\"BILLING\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"addresses\"}"))
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":12,\"customer_id\":1002,\"street\":\"281 Riverside Drive\",\"city\":\"Augusta\",\"state\":\"Georgia\",\"zip\":\"30901\",\"type\":\"BILLING\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"addresses\"}"))
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":13,\"customer_id\":1003,\"street\":\"3787 Brownton Road\",\"city\":\"Columbus\",\"state\":\"Mississippi\",\"zip\":\"39701\",\"type\":\"SHIPPING\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"addresses\"}"))
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":14,\"customer_id\":1003,\"street\":\"2458 Lost Creek Road\",\"city\":\"Bethlehem\",\"state\":\"Pennsylvania\",\"zip\":\"18018\",\"type\":\"SHIPPING\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"addresses\"}"))
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":15,\"customer_id\":1003,\"street\":\"4800 Simpson Square\",\"city\":\"Hillsdale\",\"state\":\"Oklahoma\",\"zip\":\"73743\",\"type\":\"BILLING\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"addresses\"}"))
.assertNext((message) -> assertThat(payloadString(message)).isEqualTo(
.assertNext((payload) -> assertThat(payload).isEqualTo(
"{\"id\":16,\"customer_id\":1004,\"street\":\"1289 University Hill Road\",\"city\":\"Canehill\",\"state\":\"Arkansas\",\"zip\":\"72717\",\"type\":\"LIVING\",\"__deleted\":\"false\",\"__name\":\"my-topic\",\"__db\":\"inventory\",\"__op\":\"r\",\"__table\":\"addresses\"}"))
.thenCancel()
.verify();
.verify(Duration.ofSeconds(30));

}

private String payloadString(Message<?> message) {
return new String((byte[]) message.getPayload());
}

@SpringBootApplication(exclude = { MongoAutoConfiguration.class })
@Import({ TestJdbcTemplateConfiguration.class })
static class DebeziumSupplierTestApplication {
Expand Down
5 changes: 3 additions & 2 deletions supplier/spring-debezium-supplier/src/test/resources/logback-test.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,11 @@
</encoder>
</appender>

<root level="info">
<root level="warn">
<appender-ref ref="STDOUT"/>
</root>

<logger name="org.testcontainers" level="INFO"/>
<logger name="org.testcontainers" level="WARN"/>
<logger name="com.github.dockerjava" level="WARN"/>
<logger name="io.debezium" level="INFO"/>
</configuration>
Loading