mosfet.io

Debezium Articles

Fri, 21 Mar 2025 00:00:00 +0000

Build a superfast CDC with Debezium and Quarkus

Fri, 07 Mar 2025 00:00:00 +0000

Change Data Capture (CDC) is widely used in various contexts, such as microservices communication, legacy system modernization, and cache invalidation. The core idea of this pattern is to detect and track changes in a data source (e.g., a database) and propagate them to other systems in real-time or near real-time.

Debezium is a CDC platform that provides a wide range of connectors for most data sources. Beyond capturing changes, it also offers transformation capabilities through an intuitive UI for defining debezium instances. One way to use Debezium is by embedding it in a web application along with a connector to listen for changes in a data source. What’s particularly interesting is that, with some tweaks, it’s possible to build a CDC system with ultra-fast startup times and a minimal memory footprint by leveraging Quarkus and GraalVM.

Why

I’d say, why not? From another perspective, the CDC pattern operates within the realm of distributed systems, which must adhere to principles like scalability, fault tolerance, and high availability. In many cases, it’s necessary to run a separate Debezium instance for each type of event being tracked. Optimizing resource usage not only lowers costs but also helps reduce your carbon footprint.

Disclaimer

The tweaks suggested here serve solely as a proof of concept to demonstrate Debezium’s capabilities. They are not intended for production use but rather to showcase its potential. Debezium and its surrounding third-party libraries are not designed to be fully native for all use cases. Instead, the tweaks presented here focus on a specific scenario.

The Scenario

We will ingest data from a postgres database into Quarkus application with Debezium.

The side effect

This guide can also help you understand how to make a Java application native. However, the best and most reliable approach is to use only dependencies provided by Quarkus.

Prerequisite

GraalVM installed and configured appropriately (sdkman: 23.0.2-graal)
Apache Maven 3.9.9
A working container runtime (Docker or Podman)
A working C development environment

You can use this link to set up your working environment.

Some Considerations about GraalVM

GraalVM compiles Java applications into native binaries using Ahead-of-Time (AOT) compilation. This process involves analyzing the entire application to identify all reachable code, including classes, methods, and dependencies. However, one widely used Java feature —reflection— introduces dynamic behavior that cannot always be resolved at build time, often leading to runtime failures such as ClassNotFoundException. Additionally, many libraries rely on dynamic proxy generation, which is not supported in AOT compilation. This is where Quarkus comes to the rescue, providing a set of native-ready extensions and libraries to overcome these limitations.

Build the Basic Application

Alright, let’s start building the application! All the proposed code is available on GitHub (branch native for the final result, main for the debezium quick-start).

We’ll begin by updating your pom.xml, adding the necessary Quarkus and Debezium dependencies. Additionally, the Quarkus Maven plugin is required to build the application.

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>io.mosfet</groupId>
    <artifactId>debezium-native</artifactId>
    <version>1.0-SNAPSHOT</version>
    
    <properties>
        <maven.compiler.release>23</maven.compiler.release>
        <quarkus.platform.version>3.19.2</quarkus.platform.version>
        <version.debezium>3.0.8.Final</version.debezium>
    </properties>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>io.quarkus.platform</groupId>
                <artifactId>quarkus-bom</artifactId>
                <version>${quarkus.platform.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <dependency>
            <groupId>io.quarkus</groupId>
            <artifactId>quarkus-arc</artifactId>
        </dependency>
        <dependency>
            <groupId>io.debezium</groupId>
            <artifactId>debezium-embedded</artifactId>
            <version>${version.debezium}</version>
        </dependency>
        <dependency>
            <groupId>io.debezium</groupId>
            <artifactId>debezium-connector-postgres</artifactId>
            <version>${version.debezium}</version>
        </dependency>
        <dependency>
            <groupId>io.debezium</groupId>
            <artifactId>debezium-api</artifactId>
            <version>${version.debezium}</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>${quarkus.platform.group-id}</groupId>
                <artifactId>quarkus-maven-plugin</artifactId>
                <version>${quarkus.platform.version}</version>
                <extensions>true</extensions>
                <executions>
                    <execution>
                        <goals>
                            <goal>build</goal>
                            <goal>generate-code</goal>
                            <goal>generate-code-tests</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

As you can see we added the debezium-connector-postgres to track a table in postgres.

To instrument our application, we need to inject configurations while avoiding pitfalls that GraalVM doesn’t handle well (keeping in mind our initial considerations). Instead, we’ll follow the Quarkus approach by reading configurations from application.properties.

To achieve this, we can create an interface using @ConfigMapping and @StaticInitSafe, ensuring compatibility with GraalVM.

@StaticInitSafe
@ConfigMapping(prefix = "debezium")
public interface DebeziumConfiguration {
    Map<String, String> configuration();
}

The prefix is used to identify all the properties with debezium like these one:

# Debezium CDC configuration
debezium.configuration.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.configuration.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore
debezium.configuration.name=native
debezium.configuration.database.hostname=localhost
debezium.configuration.database.port=5432
debezium.configuration.database.user=postgresuser
debezium.configuration.database.password=postgrespw
debezium.configuration.database.dbname=postgresuser
debezium.configuration.topic.prefix=dbserver1
debezium.configuration.table.include.list=inventory.products
debezium.configuration.plugin.name=pgoutput
debezium.configuration.snapshot.mode=never

So now we are able to create our listener DatabaseChangeEventListener that tracks events coming from the table products inside the inventory database:

@ApplicationScoped
@Startup
public class DatabaseChangeEventListener {

    private static final Logger LOG = LoggerFactory.getLogger(DatabaseChangeEventListener.class);

    private DebeziumEngine<?> engine;
    @Inject
    private ExecutorService executor;

    @Inject
    private DebeziumConfiguration debeziumConfiguration;

    @PostConstruct
    public void startEmbeddedEngine() {
        LOG.info("Launching Debezium embedded engine");

        final Configuration config = Configuration.empty()
                .withSystemProperties(Function.identity())
                .edit()
                .with(Configuration.from(debeziumConfiguration.configuration()))
                .build();

        this.engine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(config.asProperties())
                .notifying((events, committer) -> {
                    for (RecordChangeEvent<SourceRecord> record : events) {
                        handleDbChangeEvent(record.record());
                        committer.markProcessed(record);
                    }
                    committer.markBatchFinished();
                })
                .build();

        executor.execute(engine);
    }

    @PreDestroy
    public void shutdownEngine() throws Exception {
        LOG.info("Stopping Debezium embedded engine");
        engine.close();
        executor.shutdown();
    }

    private void handleDbChangeEvent(SourceRecord record) {
        LOG.info("DB change event {}", record);

        if (record.topic().equals("dbserver1.inventory.products")) {
            Integer productId = ((Struct) record.key()).getInt32("id");
            Struct payload = (Struct) record.value();
            Operation op = Operation.forCode(payload.getString("op"));
            Long txId = ((Struct) payload.get("source")).getInt64("txId");

            LOG.info("received event with productId: {} op: {} txId: {}", productId, op, txId);
        }
    }
}

We should be able to start our application with:

mvn clean compile quarkus:dev

Obviously before start the application, you should start a postgres instance based on the application.properties. You can use for simplicity this docker compose configuration:

services:
  postgres:
    image: quay.io/debezium/example-postgres:3.0
    ports:
     - 5432:5432
    environment:
     - POSTGRES_USER=postgresuser
     - POSTGRES_PASSWORD=postgrespw

Now you should have a working application with debezium that runs inside a jvm.

Be Native

Ok now let’s deep dive into the GraalVM capabilities. You can define a maven profile native that can trigger the build:

<profiles>
    <profile>
        <id>native</id>
        <activation>
            <property>
                <name>native</name>
            </property>
        </activation>
        <properties>
            <quarkus.native.enabled>true</quarkus.native.enabled>
            <quarkus.package.jar.enabled>false</quarkus.package.jar.enabled>
        </properties>
    </profile>
</profiles>

We can try to build natively with:

mvn clean install -Dnative

as I said, you can try…but you will fail with something like that:

Fatal error: com.oracle.graal.pointsto.util.AnalysisError$ParsingError: Error encountered while parsing org.glassfish.jersey.server.wadl.WadlFeature.configure(WadlFeature.java:49) 
Parsing context:
   at root method.(Unknown Source)

Well, I never mentioned glassfish and jersey (org.glassfish.jersey.server), so why something like that?

We should analyze the dependency tree to understand what is inside our application. So we can execute:

mvn dependency:tree

and see something like:

[INFO] ---------------------< io.mosfet:debezium-native >----------------------
[INFO] Building debezium-native 1.0-SNAPSHOT
[INFO]   from pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[INFO] 
[INFO] --- dependency:3.7.0:tree (default-cli) @ debezium-native ---
[INFO] io.mosfet:debezium-native:jar:1.0-SNAPSHOT
[INFO] +- io.quarkus:quarkus-arc:jar:3.19.2:compile
[INFO] |  +- io.quarkus.arc:arc:jar:3.19.2:compile
[INFO] |  |  +- jakarta.enterprise:jakarta.enterprise.cdi-api:jar:4.1.0:compile
[INFO] |  |  |  +- jakarta.enterprise:jakarta.enterprise.lang-model:jar:4.1.0:compile
[INFO] |  |  |  +- jakarta.el:jakarta.el-api:jar:5.0.1:compile
[INFO] |  |  |  \- jakarta.interceptor:jakarta.interceptor-api:jar:2.2.0:compile
[INFO] |  |  +- jakarta.annotation:jakarta.annotation-api:jar:3.0.0:compile
[INFO] |  |  +- jakarta.transaction:jakarta.transaction-api:jar:2.0.1:compile
[INFO] |  |  +- io.smallrye.reactive:mutiny:jar:2.8.0:compile
[INFO] |  |  |  +- io.smallrye.common:smallrye-common-annotation:jar:2.10.0:compile
[INFO] |  |  |  \- org.jctools:jctools-core:jar:4.0.5:compile
[INFO] |  |  \- org.jboss.logging:jboss-logging:jar:3.6.1.Final:compile
[INFO] |  +- io.quarkus:quarkus-core:jar:3.19.2:compile
[INFO] |  |  +- jakarta.inject:jakarta.inject-api:jar:2.0.1:compile
[INFO] |  |  +- io.smallrye.common:smallrye-common-os:jar:2.10.0:compile
[INFO] |  |  +- io.quarkus:quarkus-ide-launcher:jar:3.19.2:compile
[INFO] |  |  +- io.quarkus:quarkus-development-mode-spi:jar:3.19.2:compile
[INFO] |  |  +- io.smallrye.config:smallrye-config:jar:3.11.4:compile
[INFO] |  |  |  \- io.smallrye.config:smallrye-config-core:jar:3.11.4:compile
[INFO] |  |  |     +- org.eclipse.microprofile.config:microprofile-config-api:jar:3.1:compile
[INFO] |  |  |     +- io.smallrye.common:smallrye-common-classloader:jar:2.10.0:compile
[INFO] |  |  |     \- io.smallrye.config:smallrye-config-common:jar:3.11.4:compile
[INFO] |  |  +- org.jboss.logmanager:jboss-logmanager:jar:3.1.1.Final:compile
[INFO] |  |  |  +- io.smallrye.common:smallrye-common-constraint:jar:2.10.0:compile
[INFO] |  |  |  +- io.smallrye.common:smallrye-common-cpu:jar:2.10.0:compile
[INFO] |  |  |  +- io.smallrye.common:smallrye-common-expression:jar:2.10.0:compile
[INFO] |  |  |  +- io.smallrye.common:smallrye-common-net:jar:2.10.0:compile
[INFO] |  |  |  +- io.smallrye.common:smallrye-common-ref:jar:2.10.0:compile
[INFO] |  |  |  +- jakarta.json:jakarta.json-api:jar:2.1.3:compile
[INFO] |  |  |  \- org.eclipse.parsson:parsson:jar:1.1.7:compile
[INFO] |  |  +- org.jboss.logging:jboss-logging-annotations:jar:3.0.4.Final:compile
[INFO] |  |  +- org.jboss.threads:jboss-threads:jar:3.8.0.Final:compile
[INFO] |  |  |  \- io.smallrye.common:smallrye-common-function:jar:2.10.0:compile
[INFO] |  |  +- org.jboss.slf4j:slf4j-jboss-logmanager:jar:2.0.0.Final:compile
[INFO] |  |  +- org.wildfly.common:wildfly-common:jar:2.0.1:compile
[INFO] |  |  +- io.quarkus:quarkus-bootstrap-runner:jar:3.19.2:compile
[INFO] |  |  |  +- io.quarkus:quarkus-classloader-commons:jar:3.19.2:compile
[INFO] |  |  |  +- io.smallrye.common:smallrye-common-io:jar:2.10.0:compile
[INFO] |  |  |  \- io.github.crac:org-crac:jar:0.1.3:compile
[INFO] |  |  \- io.quarkus:quarkus-fs-util:jar:0.0.10:compile
[INFO] |  \- org.eclipse.microprofile.context-propagation:microprofile-context-propagation-api:jar:1.3:compile
[INFO] +- io.debezium:debezium-embedded:jar:3.0.8.Final:compile
[INFO] |  +- io.debezium:debezium-core:jar:3.0.8.Final:compile
[INFO] |  |  +- com.fasterxml.jackson.core:jackson-core:jar:2.18.2:compile
[INFO] |  |  +- com.fasterxml.jackson.core:jackson-databind:jar:2.18.2:compile
[INFO] |  |  \- com.fasterxml.jackson.datatype:jackson-datatype-jsr310:jar:2.18.2:compile
[INFO] |  +- org.slf4j:slf4j-api:jar:2.0.6:compile
[INFO] |  +- org.apache.kafka:connect-api:jar:3.9.0:compile
[INFO] |  |  +- org.apache.kafka:kafka-clients:jar:3.9.0:compile
[INFO] |  |  |  +- com.github.luben:zstd-jni:jar:1.5.6-4:runtime
[INFO] |  |  |  +- org.lz4:lz4-java:jar:1.8.0:runtime
[INFO] |  |  |  \- org.xerial.snappy:snappy-java:jar:1.1.10.5:runtime
[INFO] |  |  \- javax.ws.rs:javax.ws.rs-api:jar:2.1.1:runtime
[INFO] |  +- org.apache.kafka:connect-runtime:jar:3.9.0:compile
[INFO] |  |  +- org.apache.kafka:connect-transforms:jar:3.9.0:compile
[INFO] |  |  +- ch.qos.reload4j:reload4j:jar:1.2.25:runtime
[INFO] |  |  +- org.bitbucket.b_c:jose4j:jar:0.9.6:runtime
[INFO] |  |  +- com.fasterxml.jackson.core:jackson-annotations:jar:2.18.2:compile
[INFO] |  |  +- com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:jar:2.18.2:runtime
[INFO] |  |  |  +- com.fasterxml.jackson.jaxrs:jackson-jaxrs-base:jar:2.18.2:runtime
[INFO] |  |  |  \- com.fasterxml.jackson.module:jackson-module-jaxb-annotations:jar:2.18.2:runtime
[INFO] |  |  |     \- jakarta.activation:jakarta.activation-api:jar:2.1.3:runtime
[INFO] |  |  +- org.glassfish.jersey.containers:jersey-container-servlet:jar:2.39.1:runtime
[INFO] |  |  |  +- org.glassfish.jersey.containers:jersey-container-servlet-core:jar:2.39.1:runtime
[INFO] |  |  |  |  \- org.glassfish.hk2.external:jakarta.inject:jar:2.6.1:runtime
[INFO] |  |  |  +- org.glassfish.jersey.core:jersey-common:jar:2.39.1:runtime
[INFO] |  |  |  |  \- org.glassfish.hk2:osgi-resource-locator:jar:1.0.3:runtime
[INFO] |  |  |  +- org.glassfish.jersey.core:jersey-server:jar:2.39.1:runtime
[INFO] |  |  |  |  +- org.glassfish.jersey.core:jersey-client:jar:2.39.1:runtime
[INFO] |  |  |  |  \- jakarta.validation:jakarta.validation-api:jar:3.0.2:runtime
[INFO] |  |  |  \- jakarta.ws.rs:jakarta.ws.rs-api:jar:3.1.0:runtime
[INFO] |  |  +- org.glassfish.jersey.inject:jersey-hk2:jar:2.39.1:runtime
[INFO] |  |  |  +- org.glassfish.hk2:hk2-locator:jar:2.6.1:runtime
[INFO] |  |  |  |  +- org.glassfish.hk2.external:aopalliance-repackaged:jar:2.6.1:runtime
[INFO] |  |  |  |  +- org.glassfish.hk2:hk2-api:jar:2.6.1:runtime
[INFO] |  |  |  |  \- org.glassfish.hk2:hk2-utils:jar:2.6.1:runtime
[INFO] |  |  |  \- org.javassist:javassist:jar:3.29.0-GA:runtime
[INFO] |  |  +- javax.xml.bind:jaxb-api:jar:2.3.1:runtime
[INFO] |  |  |  \- javax.activation:javax.activation-api:jar:1.2.0:runtime
[INFO] |  |  +- javax.activation:activation:jar:1.1.1:runtime
[INFO] |  |  +- org.eclipse.jetty:jetty-server:jar:9.4.56.v20240826:runtime
[INFO] |  |  |  +- javax.servlet:javax.servlet-api:jar:3.1.0:runtime
[INFO] |  |  |  +- org.eclipse.jetty:jetty-http:jar:9.4.56.v20240826:runtime
[INFO] |  |  |  \- org.eclipse.jetty:jetty-io:jar:9.4.56.v20240826:runtime
[INFO] |  |  +- org.eclipse.jetty:jetty-servlet:jar:9.4.56.v20240826:runtime
[INFO] |  |  |  +- org.eclipse.jetty:jetty-security:jar:9.4.56.v20240826:runtime
[INFO] |  |  |  \- org.eclipse.jetty:jetty-util-ajax:jar:9.4.56.v20240826:runtime
[INFO] |  |  +- org.eclipse.jetty:jetty-servlets:jar:9.4.56.v20240826:runtime
[INFO] |  |  |  +- org.eclipse.jetty:jetty-continuation:jar:9.4.56.v20240826:runtime
[INFO] |  |  |  \- org.eclipse.jetty:jetty-util:jar:9.4.56.v20240826:runtime
[INFO] |  |  +- org.eclipse.jetty:jetty-client:jar:9.4.56.v20240826:runtime
[INFO] |  |  +- org.reflections:reflections:jar:0.10.2:runtime
[INFO] |  |  |  \- com.google.code.findbugs:jsr305:jar:3.0.2:runtime
[INFO] |  |  +- org.apache.maven:maven-artifact:jar:3.9.9:runtime
[INFO] |  |  |  \- org.codehaus.plexus:plexus-utils:jar:3.5.1:runtime
[INFO] |  |  \- io.swagger.core.v3:swagger-annotations:jar:2.2.8:runtime
[INFO] |  +- org.apache.kafka:connect-json:jar:3.9.0:compile
[INFO] |  |  +- com.fasterxml.jackson.datatype:jackson-datatype-jdk8:jar:2.18.2:compile
[INFO] |  |  \- com.fasterxml.jackson.module:jackson-module-afterburner:jar:2.18.2:compile
[INFO] |  \- org.apache.kafka:connect-file:jar:3.9.0:compile
[INFO] +- io.debezium:debezium-connector-postgres:jar:3.0.8.Final:compile
[INFO] |  +- org.postgresql:postgresql:jar:42.7.5:compile
[INFO] |  \- com.google.protobuf:protobuf-java:jar:3.25.5:compile
[INFO] \- io.debezium:debezium-api:jar:3.0.8.Final:compile

If you take a look inside kafka:connect-runtime, you’ll find Jetty, Jersey, and many other components that GraalVM doesn’t handle well.

Why does kafka:connect-runtime include all these dependencies? Well, I assume this library is designed to run a Kafka Connect node, but in our specific case —Debezium in a Quarkus application— this is unnecessary (most likely, Debezium only uses certain Kafka classes for data transformation).

To minimize issues with the native build, I recommend excluding all unused dependencies. Additionally, if we dig deeper into the dependency tree, we’ll find other potential troublemakers like Jackson and PostgreSQL drivers, which could cause further complications. I suggest excluding them as well and replacing them with Quarkus-recommended alternatives.

Here’s a cleaned-up pom.xml dependencies section:

    <dependencies>
        <dependency>
            <groupId>io.quarkus</groupId>
            <artifactId>quarkus-arc</artifactId>
        </dependency>
        <dependency>
            <groupId>io.debezium</groupId>
            <artifactId>debezium-embedded</artifactId>
            <version>${version.debezium}</version>
            <exclusions>
                <exclusion>
                    <groupId>com.fasterxml.jackson.core</groupId>
                    <artifactId>jackson-core</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.fasterxml.jackson.core</groupId>
                    <artifactId>jackson-databind</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.fasterxml.jackson.datatype</groupId>
                    <artifactId>jackson-datatype-jsr310</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.fasterxml</groupId>
                    <artifactId>jackson-module-afterburner</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>connect-json</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>connect-runtime</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>io.debezium</groupId>
            <artifactId>debezium-connector-postgres</artifactId>
            <version>${version.debezium}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.postgresql</groupId>
                    <artifactId>postgresql</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>io.debezium</groupId>
            <artifactId>debezium-api</artifactId>
            <version>${version.debezium}</version>
        </dependency>

        <!-- kafka runtime & connect-json (used by debezium engine) excluded
        of unnecessary or harmful for native build dependencies -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>connect-runtime</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>com.fasterxml.jackson.core</groupId>
                    <artifactId>jackson-annotations</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.bitbucket.b_c</groupId>
                    <artifactId>jose4j</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.fasterxml.jackson.jaxrs</groupId>
                    <artifactId>jackson-jaxrs-json-provider</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.glassfish.jersey.containers</groupId>
                    <artifactId>jersey-container-servlet</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.glassfish.jersey.inject</groupId>
                    <artifactId>jersey-hk2</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>javax.xml.bind</groupId>
                    <artifactId>jaxb-api</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>javax.activation</groupId>
                    <artifactId>activation</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.eclipse.jetty</groupId>
                    <artifactId>jetty-server</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.eclipse.jetty</groupId>
                    <artifactId>jetty-servlet</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.eclipse.jetty</groupId>
                    <artifactId>jetty-servlets</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.eclipse.jetty</groupId>
                    <artifactId>jetty-client</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.eclipse.jetty</groupId>
                    <artifactId>jetty-client</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.reflections</groupId>
                    <artifactId>reflections</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.maven</groupId>
                    <artifactId>maven-artifact</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>connect-json</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>com.fasterxml</groupId>
                    <artifactId>jackson-module-afterburner</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <!-- add alternatives for jackson & postgres native friendly -->
        <dependency>
            <groupId>io.quarkus</groupId>
            <artifactId>quarkus-jackson</artifactId>
        </dependency>
        <dependency>
            <groupId>io.quarkus</groupId>
            <artifactId>quarkus-jdbc-postgresql</artifactId>
        </dependency>

        <!-- in some cases kafka use jetty-util and connect-transforms -->
        <dependency>
            <groupId>org.eclipse.jetty</groupId>
            <artifactId>jetty-util</artifactId>
            <version>9.4.56.v20240826</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>connect-transforms</artifactId>
        </dependency>
    </dependencies>

The build will be successful (probably…)! So now you can run the native application (supposing that you are using Gnu/Linux or MacOS):

./target/debezium-native-1.0-SNAPSHOT-runner

but now the things are even worse. You have a runtime error like:

2025-03-11 20:08:34,408 ERROR [io.qua.run.Application] (main) Failed to start application: java.lang.RuntimeException: Failed to start quarkus
        at io.quarkus.runner.ApplicationImpl.doStart(Unknown Source)
        at io.quarkus.runtime.Application.start(Application.java:101)
        at io.quarkus.runtime.ApplicationLifecycleManager.run(ApplicationLifecycleManager.java:119)
        at io.quarkus.runtime.Quarkus.run(Quarkus.java:71)
        at io.quarkus.runtime.Quarkus.run(Quarkus.java:44)
        at io.quarkus.runtime.Quarkus.run(Quarkus.java:124)
        at io.quarkus.runner.GeneratedMain.main(Unknown Source)
Caused by: io.debezium.DebeziumException: No implementation of Debezium engine builder was found

Debezium use some reflection magic in order to instrument the engine. To fix it, we should instrument our application to register the ConvertingEngineBuilderFactory.class inside the application adding the following class:

@RegisterForReflection(targets = { ConvertingEngineBuilderFactory.class })
public class ReflectingConfig { }

If you are still with me, you will have now again an error at build time with something like:

Fatal error: com.oracle.graal.pointsto.constraints.UnsupportedFeatureException: Detected an instance of Random/SplittableRandom class in the image heap. Instances created during image generation have cached seed values and don't behave as expected. If these objects should not be stored in the image heap, you can use 

    '--trace-object-instantiation=java.util.Random'

Random and SplittableRandom are meant to provide random values and are typically expected to get a fresh seed in each run. Using a native image can open the risk to have cached values breaking the expectation. This Random definition is inside kafka libraries in a static way. To fix this kind of issue, we should instrument the application to accept at runtime the Random in this way:

# Kafka-clients  use Random (SaslClientAuthenticator line 63)
quarkus.native.additional-build-args=--initialize-at-run-time=org.apache.kafka.common.security.authenticator.SaslClientAuthenticator

The build should be successful but again, you will have an error at runtime. Now the process is the same applied to ConvertingEngineBuilderFactory.class: instrument GraalVM to find the classes that are injected by reflection. For the sake of your mental sanity, here a ReflectingConfig with all the classes for this use case:

@RegisterForReflection(targets = { ConvertingEngineBuilderFactory.class,
        SaslClientAuthenticator.class,
        JsonConverter.class,
        PostgresConnector.class,
        PostgresSourceInfoStructMaker.class,
        DefaultTransactionMetadataFactory.class,
        SchemaTopicNamingStrategy.class,
        OffsetCommitPolicy.class,
        PostgresConnectorTask.class,
        SinkNotificationChannel.class,
        LogNotificationChannel.class,
        JmxNotificationChannel.class,
        SnapshotLock.class,
        NoLockingSupport.class,
        NoSnapshotLock.class,
        SharedSnapshotLock.class,
        SelectAllSnapshotQuery.class,
        AlwaysSnapshotter.class,
        InitialSnapshotter.class,
        InitialOnlySnapshotter.class,
        NoDataSnapshotter.class,
        RecoverySnapshotter.class,
        WhenNeededSnapshotter.class,
        NeverSnapshotter.class,
        SchemaOnlySnapshotter.class,
        SchemaOnlyRecoverySnapshotter.class,
        ConfigurationBasedSnapshotter.class,
        SourceSignalChannel.class,
        KafkaSignalChannel.class,
        FileSignalChannel.class,
        JmxSignalChannel.class,
        InProcessSignalChannel.class,
        StandardActionProvider.class
})
public class ReflectingConfig { }

If you followed the instructions, now the application should work. I didn’t have time to evaluate performances but the memory consumption dropped from 150MB to 50MB.

Considerations

This is not (and I repeat, not) something ready for production use. There is still significant work to be done in two key areas:

Decoupling Debezium from Kafka Libraries: Kafka and Debezium should share only a standardized method for moving and transforming data. The Kafka runtime used for creating Kafka Connect nodes should not be a dependency that Debezium imports unnecessarily.
Removing Reflection in Debezium: Debezium should adopt an alternative approach for injecting classes dynamically, following the best practices recommended by the Quarkus team to eliminate reliance on reflection.

Addressing these challenges could pave the way for a Quarkus Extension that fully certifies Debezium’s native compatibility.

Future Scenario

The ability to drastically reduce Debezium’s memory footprint is particularly interesting, especially when considering the scenario of running Debezium Server as a native image.

As far as I know (though not 100% certain), to ensure scalability and maintain event order, there is typically a 1:1 association between a source (such as a database table) and a Debezium engine. This drop can open many scenarios in which debezium server can be the best choice for CDC.

References

About Giovanni Panice

Thu, 25 Jan 2024 00:00:00 +0000

Hi there

My name is Giovanni, a lazy pizza fan which in the spare time is an Engineer.

I currently work as a Senior Software Engineer at Red Hat, contributing to Debezium, an open-source distributed change data capture platform.

Before jump in open-source, I had the opportunity to work on the lastminute.com pricing engine for flights across meta-search and direct channels, implementing dynamic pricing algorithms to ensure accuracy and scalability under high traffic conditions. Furthermore, I worked on integrating multiple Global Distribution Systems (GDS), building reliable and high-performance systems to enhance booking and pricing capabilities.

Prior to this experience, I worked at BMC Software on capacity planning solutions for on-premises and cloud infrastructure, handling the ingestion of large volumes of metrics analyzed by statistical algorithms for forecasting.

In a past life, I spent time in the Aerospace & Defense industry, conducting research on cyber threats for unmanned aerial vehicles (UAVs). In particular, I worked on time series analysis algorithms for navigation systems at the Italian Aerospace Research Centre and was a consultant for Agusta Westland (now Leonardo Helicopters), contributing to the development of a common platform for cockpit systems.

Beyond my technical work, I co-organize the Ticino Software Craft meetup, where we exchange knowledge, discuss best practices, and explore new perspectives on software. I believe that continuous learning and sharing ideas are key to growth in our industry.

Publications & Conferences

Author: A SVM-based detection approach for GPS spoofing attacks to UAV, IEEE
Speaker: 23rd International Conference on Automation and Computing (ICAC),2017 Huddersfield University
Attendant: Special Day on Security tutorials within the International conference on Design & Technology of Integrated System in Nanoscale Era, DTIS2015, Naples
Attendant: "The Memories of Tomorrow: Technology, Design, Test and Dependability, DTIS2015, Naples

Causes I care about:

Contact

Thu, 25 Jan 2024 00:00:00 +0000

Want to get in touch with me? Send me a mail, I’ll try to get back as soon as possible!

Consumer-Driven Contract Testing with Pact and Java - Part II

Wed, 17 Jun 2020 00:00:00 +0000

The previous post explains the principles and motivations behind contract testing. Today we give a look to how write consumer-driven contract tests with pact and Java in a SpringBoot application.

Pact foundation provides junit5 integration for creation and verification of contracts.

Let’s start!

What you need

The Java Pact Testing framework used in this example is the 4.0.0, based on v3 specification.

Furthermore, you should have:

jdk 1.8 or later
maven 3.2+
a bit of testing knowledge in Spring

All the presented code is on github.

The example

The proposed example is similar to the previous one seen in part I: we have a provider service which expose an API that given a sentence and a timestamp, it replies an echo response enriched with local timestamp. This API is consumed by another service. To summarize:

endpoint POST /api/echo

request body

{
    "timestamp": 1593373353,
    "sentence": "hello!"
}

response body

{
    "phrase": "hello! sent at: 1593373353 worked at: 1593373360"
}

Consumer side

We are driven by Consumer, so then we start working on consumer side: we add the pact maven dependency in consumer pom.xml

<dependency>
    <groupId>au.com.dius</groupId>
    <artifactId>pact-jvm-consumer-junit5</artifactId>
    <version>4.0.10</version>
</dependency>

create a contract

Let’s start creating a junit5 test with PactConsumerTestExt junit extension:


import au.com.dius.pact.consumer.junit5.PactConsumerTestExt;
import org.junit.jupiter.api.extension.ExtendWith;

@ExtendWith(PactConsumerTestExt.class)
class ConsumerContractTest {

in @BeforeEach method we can assert that the mockServer which will serve the contracts is correctly up:

    @BeforeEach
    public void setUp(MockServer mockServer) {
        assertThat(mockServer, is(notNullValue()));
    }

ok, now we can create a contract. A contract can be defined with a method annotated with @Pact that returns a RequestResponsePact and provides as parameter PactDslWithProvider. All methods annotated with @Pact are used to instrument the mock server through the PactDslWithProvider in this way:

    @Pact(provider = "providerMicroservice", consumer = "consumerMicroservice")
    public RequestResponsePact echoRequest(PactDslWithProvider builder) {

        return builder
                .given("a sentence worked at 1593373360")
                .uponReceiving("an echo request at 1593373353")
                .path(API_ECHO) /* request */
                .method("POST")
                .body(echoRequest) 
                .willRespondWith() /* response */
                .status(200)
                .headers(headers)
                .body(echoResponse)
                .toPact();
    }

The Pact DSL provides a fluent API very similar to Spring mockMvc: Here we are saying that when the mock server receives an echoRequest, it should return 200 and an echoResponse. The given and the uponReceiving method, define the specification in bdd approach and the Pact testing framework, uses the given part to bring the provider into the correct state before executing the interaction defined in the contract.

Matchers: build a response with PactDslJsonBody

in the previous step we created an interaction using two json object(echoRequest and echoResponse). On the provider side, the test verify that the generated response is perfectly equal to the one defined in the contract.

The Pact testing framework provides also a DSL that permits the definition of different matching case in this way:

    @Pact(provider = "providerMicroservice", consumer = "consumerMicroservice")
    public RequestResponsePact echoRequestWithDsl(PactDslWithProvider builder) {

        PactDslJsonBody responseWrittenWithDsl = new PactDslJsonBody()
                .stringType("phrase", "hello! sent at: X worked at: Y") /* match on type */
                .close()
                .asBody();

        return builder
                .given("WITH DSL: a sentence worked at 1593373360")
                .uponReceiving("an echo request at 1593373353")
                .path(API_ECHO)
                .method("POST")
                .body(echoRequest)
                .willRespondWith()
                .status(200)
                .headers(headers)
                .body(responseWrittenWithDsl)
                .toPact();
    }

Here we created a response with PactDslJsonBody DSL that defines a match case based on type instead of value. It’s possible with PactDslJsonBody different match case based on regex or array length.

Verify the contract

Now we can create the real test which verify the contract on the consumer side:

    @Test
    @PactTestFor(pactMethod = "echoRequest")
    @DisplayName("given a sentence with a timestamp, when calling producer microservice, than I receive back an echo sentence with a timestamp")
    void givenASentenceWithATimestampWhenCallingProducerThanReturnAnEchoWithATimestamp(MockServer mockServer) throws IOException {
        
        BasicHttpEntity bodyRequest = new BasicHttpEntity();
        bodyRequest.setContent(IOUtils.toInputStream(echoRequest, Charset.defaultCharset()));

        HttpResponse httpResponse = Request.Post(mockServer.getUrl() + API_ECHO)
                .body(bodyRequest)
                .execute()
                .returnResponse();

        ObjectMapper objectMapper = new ObjectMapper();
        EchoResponse actualResult = objectMapper.readValue(httpResponse.getEntity().getContent(), EchoResponse.class);
        
        assertEquals(expectedResult, actualResult);
    }

if we run mvn test and we don’t have errors, we will see in ./target/pacts a json file that use the pact formalism for contracts. We use the generated contract in the provider-side.

Provider side

For the provider, we have a different dependency to add in pom.xml:

		<dependency>
			<groupId>au.com.dius</groupId>
			<artifactId>pact-jvm-provider-junit5</artifactId>
			<version>4.0.10</version>
		</dependency>

Verify the contract on provider side

Here is the thing: we need to verify the contract against provider implementation. In the Spring world, it’s sounds like an integration test which verify the web layer. So here the magic:

@ExtendWith(SpringExtension.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT, classes = ProviderApplication.class)
@EnableAutoConfiguration
@AutoConfigureMockMvc
@TestPropertySource(locations = "classpath:application-contract-test.properties")
@Provider("providerMicroservice")
@PactFolder("../consumer/target/pacts")
public class ProviderContractTest {
    
    @Value("${server.host}")
    private String serverHost;
    @Value("${server.port}")
    private int serverPort;

    @BeforeEach
    void setupTestTarget(PactVerificationContext context) {
        context.setTarget(new HttpTestTarget(serverHost, serverPort, "/"));
    }

    @TestTemplate
    @ExtendWith(PactVerificationInvocationContextProvider.class)
    void pactVerificationTestTemplate(PactVerificationContext context) {
        context.verifyInteraction();
    }

    @State("a sentence worked at 1593373360")
    public void sentenceWorkedAt1593373360() {
        when(phraseService.echo(1593373353, "hello!"))
                .thenReturn(new Phrase("hello! sent at: 1593373353 worked at: 1593373360"));
    }

}

That’s it. As you can see, we have a @SpringBootTest with a fixed port and a @TestPropertySource that defines it in order to attach the pact context to the application context with host and port info. Obviously there are other ways, like random ports and so on, but the main thing here is to bind both context together.

Another thing here is the @PactFolder annotation that points to contracts generated by the consumer. The Pact Framework search for contracts that belong to the service, and run the verification.

The `@State` annotation

    @State("a sentence worked at 1593373360")
    public void sentenceWorkedAt1593373360() {
        when(phraseService.echo(1593373353, "hello!"))
                .thenReturn(new Phrase("hello! sent at: 1593373353 worked at: 1593373360"));
    }

As previously mentioned, the given statement in the consumer contract, define with a business expression, the state in which the system-under-test, should be during the execution. Following this approach, we define in the provider, a method with @state annotation, that contains the commands necessary for the correct execution. In our case, we mock the business service delegated to execute the eco logic. The framework executes the state method before calling the API defined in the contracts. The real test, in this way, is “reduced” to a simple call:


    @TestTemplate
    @ExtendWith(PactVerificationInvocationContextProvider.class)
    void pactVerificationTestTemplate(PactVerificationContext context) {
        context.verifyInteraction();
    }

Use a broker

If you have a broker that stores the contracts, you can change the @PactFolder annotation with @PactBroker one and define the following plugin in the pom.xml:

    <build>
        <plugins>
            <plugin>
                <groupId>au.com.dius</groupId>
                <artifactId>pact-jvm-provider-maven</artifactId>
                <version>4.0.0</version>
                <configuration>
                    <pactDirectory>target/pacts</pactDirectory>
                    <pactBrokerUrl>${pact.broker.protocol}://${pact.broker.host}</pactBrokerUrl>
                    <projectVersion>${contracts.version}</projectVersion>
                    <trimSnapshot>true</trimSnapshot>
                </configuration>
            </plugin>
        </plugins>
    </build>

What’s Next

We have covered how to develop and verify a simple contract with java starting from the consumer. We have used the Pact DSL and matchers introduced in v3 spec which is an interesting feature during design & testing.

In the next posts(I hope), we will see how deploy a broker server to store contracts and how integrate the entire flow with a CI.

Consumer-Driven Contract Testing - Part I

Wed, 03 Jun 2020 00:00:00 +0000

Introduction

This is the first of a series of blog posts about Contract Testing which cover the minimum set of theory and practice necessary for an effective adoption in your team, from design to code integration.

Contract Testing is a category of testing activity where the data formats and conventions defined by two systems (services) which communicate a business value, is tested against a Mock called “Contract”. A service provides a callable API that can be consumed by another (or many) service which create an interaction between parties that needs to be satisfied during the evolution and developing of services which are now coupled. The interaction between services probably rely on a communication layer which can be slow or not reachable which can affect the test results. For this reason, sometimes the best solution is to verify the interaction with a TestDouble which describe the expectations between the parties.

For the sake of clarity, I have strong opinion.

Context

Let’s suppose that we have a Product Service which has an HTTP method that provides product information in a JSON format upon receiving a product id:

HTTP GET: api/v1/product?id=123

{
  "name": "nexus",
  "type": "smartphone",
  "price": "21.03"
}

The product information are consumed by Customer Service.

Conventions are defined in this way:

Customer Service is the Consumer
Product Service is the Provider

During developing, the Product service can evolve the API in a way that the HTTP response change the data format. For example, the type field from a string value change to object:

{
  "name": "nexus",
  "type": {
    "category": "smartphone",
    "weight": 0.2,
    "color": "blue"
  },
  "price": "21.03"
}

this is defined as a breaking change in which Provider doesn’t respect the defined interaction between Consumer.

In the same way, the Customer service can evolve the request call. For example, instead of call the API with a query parameter id, it can use a field in the header to query the Product Service. Also, this example is a sort of breaking change in which Consumer doesn’t respect the defined interaction between Provider.

Therefore, the interaction between services can be broken in many ways, and the cause can be triggered by both sides which makes necessary some checks to avoid regression during the development.

e2e Testing

The simplest thing to do, is to create an end-to-end test that cover the entire calls flow between Consumer Service and Provider Service.

Despite what I previously said about simplicity, e2e testing give the best confidence on software behaviour but with some problems. They are flaky tests and often fail with false positive furthermore e2e tests are hard do maintain, slow, and it’s necessary to span multiple services in testing environment with terrible slowness caused by deploy flow and environment nightmares. There are some solutions (1, 2) that can provide the right balance between simplicity and maintainability which are a step back e2e tests but often you need a staging environment to execute tests verification or add a level of complexity to remove the environment need, which personally I found with these solutions a tedious process. Moreover, all these solutions are reactive approaches which in the simplest case, doesn’t avoid the integration of code that breaks the services interactions, but they provide a sort of alerting like open a Jira defect or github issue. It’s possible to avoid code integration in case of breaks but imply an overcomplicated CI or some experimental branch.

Mocks

~~Developers~~ (I) usually dislike maintaining things which depend on environments or that needs hours to have a result. If something is faulty and slow, it’s often untrusted and consequently useless. Furthermore, works with other services means deal with other teams which are often too much busy doing new ~~bugs~~ features. So then, to avoid this annoying stuff, it’s necessary to adopt a strategy which imply a level of complexity which usually developers are used to see: Mocks.

Mocks are a type of TestDouble that define a sort of specification based on expectations and, in this particular case, mocks can substitute APIs or clients reducing in this way, parties to set up and run.

Even if the image show deployed services, there are many solutions to load part of an application which doesn’t imply to run effectively the service. however, you can simulate an interaction through a mock or stub based on Contracts. As showed before, we can have two categories of breaking change based on the side that doesn’t maintain expectations and in the same way, we can identify two categories of contracts:

Provider Contracts
Consumer Contracts

Contracts characteristics

A Provider service exposes a set of business functionalities which can be used or not by one or many consumers and ingested in different ways. A change in the provider interface can break interactions with many consumers, but it is not the same for the consumer. In the same way, a Provider contract cover completely the functionalities exposed by the service with only one definition. On the contrary, a Consumer contract can cover only a set of functionalities that are interesting for the service.

We can say that a provider contract is a closed and complete expression of business functionalities. Instead, consumer contract is an open and incomplete definition of business expectations. When a provider accepts the expectations defined by the consumer contract, confirms that the expectation is a functionality that supported for a period of time.

Consumer-driven Contracts

Ok, so now? Well, If you take a look on internet, ~~you will not find~~ any trace of provider driven contract testing ~~or question about it~~. On the contrary, the web is full of posts about Consumer-driven contract testing. This should be enough for you and me to choose a consumer-driven solution but, as I said at the beginning, the idea of this post is to give the minimum set of knowledge about Contract testing, and to have an answer in case of your colleagues are free mind enough to ask if exists an alternative meanwhile you are exposing why Contract tests are so cool.

In short terms, it’s all about business. Consumer contracts point the finger on current supported business value exposed by the provider. In a consumer-driven approach, the sum of all contracts generated by the consumers and asserted by the provider is closed and complete respect the functionalities requested.

Anyway Consumer-driven Contracts drive the evolution of the services keeping the focus on what really matter with a feedback loop on the changes that will arrive during the life and death of the services which cannot be achievable with a Provider-driven approach which not take in account the feedback from the consumer.

In my experience, it’s easier and natural to think in a Provider-First manner than Consumer-first, and the reason behind this mindset, it might be connected to the fact that a breaking change made by the Provider is more detectable than the one made by the Consumer. In a real world example where your team isn’t the owner of both services, you have to deal with meetings, longer meetings and extravagant, informal, hermetic, long-winded design docs and other mythical beasts. A contract is a formal expression of needs and duties which can be another tool in your pocket that can be used in order to reduce the background noise during the process of evolution of services. Probably you know the sensation of powerlessness when a design doc or swagger definition reaches the mailbox or a shared folder and it’s necessary another meeting or mail tread to have a change in the definition or worse, get an HTTP response status error because someone made a little change in the API which isn’t versioned.

To summarize, Consumer-Driven contracts can give you a process that presents an iterative way of proceeding with a formalised format which can help large organization with services which are owned by different teams that can be in different locations.

Contract Testing with Pact

Pact.io is an implementation of Consumer-driven contract testing which actually support many languages. At the moment of writing this post, pact foundation released v3 specification which cover the following features:

pact format for message queues
regular expression and type matching
specification shared between Ruby, JVM and .Net versions

Furthermore, it’s available a broker which can be used to publish and share contracts between services.

What’s Next

We have covered the minimum set of principles and motivations that are necessary in my opinion to work with contract testing. In the next posts we will take a look to how:

create consumer-driven contracts tests with Pact and Java
setup broker server
integration with Continuous Integration

References

Consumer-Driven Contracts: A Service Evolution Pattern

ContractTest

TestDouble

The curious case for the Provider Driven Contract

The Practical Test Pyramid

Patterns for Managing Source COde Branches

Mocks Aren’t Stubs

7 Reasons to Choose Consumer-Driven Contract Tests Over End-to-End Tests

mosfet.io

Debezium Articles

Build a superfast CDC with Debezium and Quarkus

Why

Disclaimer

The Scenario

The side effect

Prerequisite

Some Considerations about GraalVM

Build the Basic Application

Be Native

Considerations

Future Scenario

About Giovanni Panice

Hi there

Publications & Conferences

Causes I care about:

Contact

Consumer-Driven Contract Testing with Pact and Java - Part II

What you need

The example

Consumer side

create a contract

Matchers: build a response with PactDslJsonBody

Verify the contract

Provider side

Verify the contract on provider side

The @State annotation

Use a broker

What’s Next

Consumer-Driven Contract Testing - Part I

Introduction

Context

e2e Testing

Mocks

Contracts characteristics

Consumer-driven Contracts

Contract Testing with Pact

What’s Next

References

The `@State` annotation