diff --git a/cloudevents/README.md b/cloudevents/README.md index 97ac3ba8dd..f924bd05be 100644 --- a/cloudevents/README.md +++ b/cloudevents/README.md @@ -5,7 +5,7 @@ This demo automatically deploys the topology of services as defined in the [Debe ## Preparations ```shell -export DEBEZIUM_VERSION=1.9 +export DEBEZIUM_VERSION=2.0 mvn clean install -f avro-data-extractor/pom.xml docker-compose up --build ``` @@ -51,7 +51,7 @@ Examine its contents like so: docker run --rm --tty \ --network cloudevents-network \ quay.io/debezium/tooling:1.2 \ - kafkacat -b kafka:9092 -C -o beginning -q -s value=avro -r http://schema-registry:8081 \ + kafkacat -b kafka:9092 -C -o beginning -q \ -t customers2 | jq . ``` @@ -65,7 +65,7 @@ curl -i -X PUT -H "Accept:application/json" -H "Content-Type:application/json" docker run --rm --tty \ --network cloudevents-network \ quay.io/debezium/tooling:1.2 \ - kafkacat -b kafka:9092 -C -o beginning -q -s value=avro -r http://schema-registry:8081 \ + kafkacat -b kafka:9092 -C -o beginning -q \ -t dbserver3.inventory.customers | jq . ``` @@ -76,8 +76,8 @@ The same stream processing application writes out that data to the `customers3` docker run --rm --tty \ --network cloudevents-network \ quay.io/debezium/tooling:1.2 \ - kafkacat -b kafka:9092 -C -o beginning -q -s value=avro -r http://schema-registry:8081 \ - -t customers2 | jq . + kafkacat -b kafka:9092 -C -o beginning -q \ + -t customers3 ``` ## CloudEvents Binary Mode diff --git a/cloudevents/avro-data-extractor/Dockerfile b/cloudevents/avro-data-extractor/Dockerfile deleted file mode 100644 index f1cbaafd67..0000000000 --- a/cloudevents/avro-data-extractor/Dockerfile +++ /dev/null @@ -1,5 +0,0 @@ -FROM fabric8/java-jboss-openjdk8-jdk -ENV JAVA_OPTIONS='-Dquarkus.http.host=0.0.0.0 -Dquarkus.http.port=${PORT}' -COPY target/lib/* /deployments/lib/ -COPY target/*-runner.jar /deployments/app.jar -ENTRYPOINT [ "/deployments/run-java.sh" ] diff --git a/cloudevents/avro-data-extractor/pom.xml b/cloudevents/avro-data-extractor/pom.xml index 876bb4fe67..95184b29bf 100755 --- a/cloudevents/avro-data-extractor/pom.xml +++ b/cloudevents/avro-data-extractor/pom.xml @@ -10,8 +10,8 @@ jar - 1.8 - 1.8 + 11 + 11 UTF-8 UTF-8 false @@ -21,8 +21,9 @@ 3.0.0-M6 3.2.0 - 2.7.2.Final - 1.9.5.Final + 2.13.1.Final + 2.0.0.Final + 7.2.0 @@ -76,7 +77,7 @@ io.quarkus - quarkus-undertow-websockets + quarkus-websockets io.quarkus @@ -97,7 +98,37 @@ io.confluent kafka-streams-avro-serde - 5.3.2 + 7.2.1 + + + io.apicurio + apicurio-registry-utils-serde + 1.3.2.Final + + + io.quarkus + quarkus-confluent-registry-avro + + + + io.quarkus + quarkus-rest-client-reactive + + + io.confluent + kafka-avro-serializer + ${version.kafka.avro} + + + jakarta.ws.rs + jakarta.ws.rs-api + + + + + io.confluent + kafka-connect-avro-converter + ${version.kafka.avro} diff --git a/cloudevents/avro-data-extractor/src/main/docker/Dockerfile.jvm b/cloudevents/avro-data-extractor/src/main/docker/Dockerfile.jvm new file mode 100644 index 0000000000..abfa5e5375 --- /dev/null +++ b/cloudevents/avro-data-extractor/src/main/docker/Dockerfile.jvm @@ -0,0 +1,94 @@ +#### +# This Dockerfile is used in order to build a container that runs the Quarkus application in JVM mode +# +# Before building the container image run: +# +# ./mvnw package +# +# Then, build the image with: +# +# docker build -f src/main/docker/Dockerfile.jvm -t quarkus/my-artifactId-jvm . +# +# Then run the container using: +# +# docker run -i --rm -p 8080:8080 quarkus/my-artifactId-jvm +# +# If you want to include the debug port into your docker image +# you will have to expose the debug port (default 5005) like this : EXPOSE 8080 5005 +# +# Then run the container using : +# +# docker run -i --rm -p 8080:8080 quarkus/my-artifactId-jvm +# +# This image uses the `run-java.sh` script to run the application. +# This scripts computes the command line to execute your Java application, and +# includes memory/GC tuning. +# You can configure the behavior using the following environment properties: +# - JAVA_OPTS: JVM options passed to the `java` command (example: "-verbose:class") +# - JAVA_OPTS_APPEND: User specified Java options to be appended to generated options +# in JAVA_OPTS (example: "-Dsome.property=foo") +# - JAVA_MAX_MEM_RATIO: Is used when no `-Xmx` option is given in JAVA_OPTS. This is +# used to calculate a default maximal heap memory based on a containers restriction. +# If used in a container without any memory constraints for the container then this +# option has no effect. If there is a memory constraint then `-Xmx` is set to a ratio +# of the container available memory as set here. The default is `50` which means 50% +# of the available memory is used as an upper boundary. You can skip this mechanism by +# setting this value to `0` in which case no `-Xmx` option is added. +# - JAVA_INITIAL_MEM_RATIO: Is used when no `-Xms` option is given in JAVA_OPTS. This +# is used to calculate a default initial heap memory based on the maximum heap memory. +# If used in a container without any memory constraints for the container then this +# option has no effect. If there is a memory constraint then `-Xms` is set to a ratio +# of the `-Xmx` memory as set here. The default is `25` which means 25% of the `-Xmx` +# is used as the initial heap size. You can skip this mechanism by setting this value +# to `0` in which case no `-Xms` option is added (example: "25") +# - JAVA_MAX_INITIAL_MEM: Is used when no `-Xms` option is given in JAVA_OPTS. +# This is used to calculate the maximum value of the initial heap memory. If used in +# a container without any memory constraints for the container then this option has +# no effect. If there is a memory constraint then `-Xms` is limited to the value set +# here. The default is 4096MB which means the calculated value of `-Xms` never will +# be greater than 4096MB. The value of this variable is expressed in MB (example: "4096") +# - JAVA_DIAGNOSTICS: Set this to get some diagnostics information to standard output +# when things are happening. This option, if set to true, will set +# `-XX:+UnlockDiagnosticVMOptions`. Disabled by default (example: "true"). +# - JAVA_DEBUG: If set remote debugging will be switched on. Disabled by default (example: +# true"). +# - JAVA_DEBUG_PORT: Port used for remote debugging. Defaults to 5005 (example: "8787"). +# - CONTAINER_CORE_LIMIT: A calculated core limit as described in +# https://www.kernel.org/doc/Documentation/scheduler/sched-bwc.txt. (example: "2") +# - CONTAINER_MAX_MEMORY: Memory limit given to the container (example: "1024"). +# - GC_MIN_HEAP_FREE_RATIO: Minimum percentage of heap free after GC to avoid expansion. +# (example: "20") +# - GC_MAX_HEAP_FREE_RATIO: Maximum percentage of heap free after GC to avoid shrinking. +# (example: "40") +# - GC_TIME_RATIO: Specifies the ratio of the time spent outside the garbage collection. +# (example: "4") +# - GC_ADAPTIVE_SIZE_POLICY_WEIGHT: The weighting given to the current GC time versus +# previous GC times. (example: "90") +# - GC_METASPACE_SIZE: The initial metaspace size. (example: "20") +# - GC_MAX_METASPACE_SIZE: The maximum metaspace size. (example: "100") +# - GC_CONTAINER_OPTIONS: Specify Java GC to use. The value of this variable should +# contain the necessary JRE command-line options to specify the required GC, which +# will override the default of `-XX:+UseParallelGC` (example: -XX:+UseG1GC). +# - HTTPS_PROXY: The location of the https proxy. (example: "myuser@127.0.0.1:8080") +# - HTTP_PROXY: The location of the http proxy. (example: "myuser@127.0.0.1:8080") +# - NO_PROXY: A comma separated lists of hosts, IP addresses or domains that can be +# accessed directly. (example: "foo.example.com,bar.example.com") +# +### +FROM registry.access.redhat.com/ubi8/openjdk-11:1.11 + +ENV LANGUAGE='en_US:en' + + +# We make four distinct layers so if there are application changes the library layers can be re-used +COPY --chown=185 target/quarkus-app/lib/ /deployments/lib/ +COPY --chown=185 target/quarkus-app/*.jar /deployments/ +COPY --chown=185 target/quarkus-app/app/ /deployments/app/ +COPY --chown=185 target/quarkus-app/quarkus/ /deployments/quarkus/ + +EXPOSE 8079 +USER 185 +ENV AB_JOLOKIA_OFF="" +ENV JAVA_OPTS="-Dquarkus.http.host=0.0.0.0 -Djava.util.logging.manager=org.jboss.logmanager.LogManager" +ENV JAVA_APP_JAR="/deployments/quarkus-run.jar" + diff --git a/cloudevents/avro-data-extractor/src/main/java/io/debezium/examples/cloudevents/dataextractor/StreamsPipelineManager.java b/cloudevents/avro-data-extractor/src/main/java/io/debezium/examples/cloudevents/dataextractor/StreamsPipelineManager.java index 763b120cda..48784bbba4 100644 --- a/cloudevents/avro-data-extractor/src/main/java/io/debezium/examples/cloudevents/dataextractor/StreamsPipelineManager.java +++ b/cloudevents/avro-data-extractor/src/main/java/io/debezium/examples/cloudevents/dataextractor/StreamsPipelineManager.java @@ -7,13 +7,13 @@ import java.nio.ByteBuffer; import java.util.Collections; -import java.util.Map; import javax.enterprise.context.ApplicationScoped; import javax.enterprise.inject.Produces; import javax.inject.Inject; import org.apache.avro.generic.GenericRecord; +import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsBuilder; @@ -24,10 +24,15 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig; -import io.confluent.kafka.streams.serdes.avro.GenericAvroSerde; +import io.apicurio.registry.client.CompatibleClient; +import io.apicurio.registry.client.RegistryService; +import io.apicurio.registry.utils.serde.AvroKafkaDeserializer; +import io.apicurio.registry.utils.serde.AvroKafkaSerializer; + import io.debezium.examples.cloudevents.dataextractor.model.CloudEvent; import io.debezium.serde.DebeziumSerdes; + + /** * Starts up the KStreams pipeline once the source topics have been created. * @@ -73,9 +78,10 @@ Topology createStreamTopology() { .mapValues(ce -> ce.data) .to(jsonAvroExtractedTopic, Produced.with(longKeySerde, Serdes.ByteArray())); - Serde genericAvroSerde = new GenericAvroSerde(); - Map config = Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl); - genericAvroSerde.configure(config, false); + + RegistryService service = CompatibleClient.createCompatible(schemaRegistryUrl); + Deserializer deserializer = new AvroKafkaDeserializer(service); + Serde genericAvroSerde = Serdes.serdeFrom(new AvroKafkaSerializer<>(service), deserializer); builder.stream(avroAvroCustomersTopic, Consumed.with(longKeySerde, genericAvroSerde)) .mapValues(ce -> ((ByteBuffer) ce.get("data")).array()) diff --git a/cloudevents/avro-data-extractor/src/main/java/io/debezium/examples/cloudevents/dataextractor/model/CloudEvent.java b/cloudevents/avro-data-extractor/src/main/java/io/debezium/examples/cloudevents/dataextractor/model/CloudEvent.java index ec847ef254..f633eb7d65 100644 --- a/cloudevents/avro-data-extractor/src/main/java/io/debezium/examples/cloudevents/dataextractor/model/CloudEvent.java +++ b/cloudevents/avro-data-extractor/src/main/java/io/debezium/examples/cloudevents/dataextractor/model/CloudEvent.java @@ -14,12 +14,16 @@ public class CloudEvent { public String iodebeziumconnector; public String iodebeziumname; public String iodebeziumtsms; - public boolean iodebeziumsnapshot; + public String iodebeziumsnapshot; public String iodebeziumdb; + public String iodebeziumsequence; public String iodebeziumschema; public String iodebeziumtable; public String iodebeziumtxId; + public String iodebeziumtxid; public String iodebeziumlsn; public String iodebeziumxmin; + public String iodebeziumtxtotalorder; + public String iodebeziumtxdatacollectionorder; public byte[] data; } diff --git a/cloudevents/avro-data-extractor/src/main/resources/application.properties b/cloudevents/avro-data-extractor/src/main/resources/application.properties index e2f4ac2f6d..bfc8efd301 100644 --- a/cloudevents/avro-data-extractor/src/main/resources/application.properties +++ b/cloudevents/avro-data-extractor/src/main/resources/application.properties @@ -4,9 +4,9 @@ json.avro.extracted.topic=customers2 avro.avro.customers.topic=dbserver3.inventory.customers avro.avro.extracted.topic=customers3 -schema.registry.url=http://schema-registry:8081 +schema.registry.url=http://schema-registry:8080/api/ -quarkus.kafka-streams.bootstrap-servers=localhost:9092 +quarkus.kafka-streams.bootstrap-servers=kafka:9092 quarkus.kafka-streams.application-id=cloudevents-data-extractor quarkus.kafka-streams.topics=${json.avro.customers.topic},${avro.avro.customers.topic} diff --git a/cloudevents/docker-compose.yaml b/cloudevents/docker-compose.yaml index 82ee8b94fd..c9914ff549 100644 --- a/cloudevents/docker-compose.yaml +++ b/cloudevents/docker-compose.yaml @@ -44,38 +44,20 @@ services: - STATUS_STORAGE_TOPIC=my_connect_statuses - KAFKA_DEBUG=true - DEBUG_SUSPEND_FLAG=n + - ENABLE_APICURIO_CONVERTERS=true networks: - my-network schema-registry: - image: confluentinc/cp-schema-registry:7.0.1 + image: apicurio/apicurio-registry-mem:2.3.1.Final ports: - - 8181:8181 - - 8081:8081 - environment: - - SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=kafka:9092 - - SCHEMA_REGISTRY_HOST_NAME=schema-registry - - SCHEMA_REGISTRY_LISTENERS=http://schema-registry:8081 - - SCHEMA_REGISTRY_ACCESS_CONTROL_ALLOW_METHODS=GET,POST,PUT,OPTIONS - - SCHEMA_REGISTRY_ACCESS_CONTROL_ALLOW_ORIGIN=* - links: - - zookeeper - networks: - - my-network - schema-registry-ui: - image: landoop/schema-registry-ui - ports: - - 8000:8000 - environment: - - SCHEMAREGISTRY_URL=http://schema-registry:8081 - - PROXY=true - links: - - schema-registry + - 8080:8080 networks: - my-network avro-extractor: image: debezium-examples/cloudevents-avro-extractor:${DEBEZIUM_VERSION} build: context: avro-data-extractor + dockerfile: src/main/docker/Dockerfile.jvm ports: - 8079:8079 links: diff --git a/cloudevents/register-postgres-avro-avro.json b/cloudevents/register-postgres-avro-avro.json index 6d38430290..39a537c390 100644 --- a/cloudevents/register-postgres-avro-avro.json +++ b/cloudevents/register-postgres-avro-avro.json @@ -6,7 +6,7 @@ "database.user": "postgres", "database.password": "postgres", "database.dbname" : "postgres", - "database.server.name": "dbserver3", + "topic.prefix": "dbserver3", "slot.name":"dbserver3", "schema.include.list": "inventory", "key.converter": "org.apache.kafka.connect.json.JsonConverter", @@ -14,5 +14,7 @@ "value.converter": "io.debezium.converters.CloudEventsConverter", "value.converter.serializer.type" : "avro", "value.converter.data.serializer.type" : "avro", - "value.converter.avro.schema.registry.url": "http://schema-registry:8081" + "value.converter.avro.apicurio.registry.url": "http://schema-registry:8080/apis/registry/v2", + "value.converter.avro.apicurio.registry.auto-register": "true", + "value.converter.avro.apicurio.registry.find-latest": "true" } diff --git a/cloudevents/register-postgres-json-avro.json b/cloudevents/register-postgres-json-avro.json index 73420132e2..0716269c7f 100644 --- a/cloudevents/register-postgres-json-avro.json +++ b/cloudevents/register-postgres-json-avro.json @@ -6,12 +6,14 @@ "database.user": "postgres", "database.password": "postgres", "database.dbname" : "postgres", - "database.server.name": "dbserver2", + "topic.prefix": "dbserver2", "slot.name":"dbserver2", "schema.include.list": "inventory", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.enable": "false", "value.converter": "io.debezium.converters.CloudEventsConverter", "value.converter.data.serializer.type" : "avro", - "value.converter.avro.schema.registry.url": "http://schema-registry:8081" + "value.converter.avro.apicurio.registry.url": "http://schema-registry:8080/apis/registry/v2", + "value.converter.avro.apicurio.registry.auto-register": "true", + "value.converter.avro.apicurio.registry.find-latest": "true" } diff --git a/cloudevents/register-postgres-json-json.json b/cloudevents/register-postgres-json-json.json index 6a8340b73f..78879cad03 100644 --- a/cloudevents/register-postgres-json-json.json +++ b/cloudevents/register-postgres-json-json.json @@ -6,7 +6,7 @@ "database.user": "postgres", "database.password": "postgres", "database.dbname" : "postgres", - "database.server.name": "dbserver1", + "topic.prefix": "dbserver1", "slot.name":"dbserver1", "schema.include.list": "inventory", "key.converter": "org.apache.kafka.connect.json.JsonConverter",