diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/README.adoc b/schema-registry-samples/schema-registry-confluent-avro-serializer/README.adoc index e5d39ad..706f4ca 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/README.adoc +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/README.adoc @@ -113,3 +113,73 @@ Registry server and Avro, the schema evolution occurs behind the scenes. The aut Once you are done with running the samples, stop the docker containers `docker-compose down` + + +==== Using Confluent Command Center + +This test can be run with the [Confluent Command Center](https://docs.confluent.io/current/control-center/index.html) - a web-based tool for managing and monitoring Apache Kafka®. Control Center facilitates building and monitoring production data pipelines and streaming applications. + +For a quick start you can install the command center with the help of the `docker-compose-control-center.yaml`: + +``` +docker-compose -f ./docker-compose-control-center.yaml +``` + +For further info check the Confluent's [Quick Start](https://docs.confluent.io/current/quickstart/ce-docker-quickstart.html). + +The [Schema Registry feature in Control Center](https://docs.confluent.io/current/control-center/topics/schema.html) would help you to visualize and manage the topic schemas. + +After you run the samples and post couple of messages as explained above. + +1. Open the command center at `http://localhost:9021` and click on the provided cluster. +2. From the vertical menu select `Topics` tab. +3. From the list of topics select the `sensor-topic` - the topic created by the samples. +4. Click on the `Schema` tab to see the `Sensors` schema. + +You can also use the Confluent Schema REST API at `http://localhost:8081`. For example the `http://localhost:8081/subjects` will list the schema names (e.g. subjects) defined. +After you have run the samples you should be able to see a schema subject name `sensor-topic-value`. + +===== NOTE + +By default Kafka uses the [TopicNameStrategy](https://docs.confluent.io/current/schema-registry/serdes-develop/index.html) to create the name of the message payload schema. Later means that the schema is named after your topic name (e.g. spring.cloud.stream.bindings.:destination) with `-value` suffix. + +That means that by default you can use a single schema per topic. The subject naming strategy can be changed to `RecordNameStrategy` or `TopicRecordNameStrategy` with the help of the `spring.cloud.stream.kafka.binder.consumerProperties` and `spring.cloud.stream.kafka.binder.producerProperties` properties like this: + +Extend your consumer configuration like this: +``` +spring: + cloud: + stream: + ......... + kafka: + binder: + consumerProperties: + value: + subject: + name: + strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy +``` + +Extend your producer configuration like this: +``` +spring: + cloud: + stream: + ......... + kafka: + binder: + producerProperties: + value: + subject: + name: + strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy +``` + +Note that currently the Command Center seams to be recognizing only the subjects created with `TopicNameStrategy` . If you configure the `RecordNameStrategy` they schema will not appear in the UI. + + + + + + + diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/docker-compose-control-center.yaml b/schema-registry-samples/schema-registry-confluent-avro-serializer/docker-compose-control-center.yaml new file mode 100644 index 0000000..85ed076 --- /dev/null +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/docker-compose-control-center.yaml @@ -0,0 +1,178 @@ +--- +version: '2' +services: + zookeeper: + image: confluentinc/cp-zookeeper:5.5.0 + hostname: zookeeper + container_name: zookeeper + ports: + - "2181:2181" + environment: + ZOOKEEPER_CLIENT_PORT: 2181 + ZOOKEEPER_TICK_TIME: 2000 + + broker: + image: confluentinc/cp-server:5.5.0 + hostname: broker + container_name: broker + depends_on: + - zookeeper + ports: + - "9092:9092" + environment: + KAFKA_BROKER_ID: 1 + KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181' + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092 + KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 + KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 + KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 + CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092 + CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper:2181 + CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1 + CONFLUENT_METRICS_ENABLE: 'true' + CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous' + + schema-registry: + image: confluentinc/cp-schema-registry:5.5.0 + hostname: schema-registry + container_name: schema-registry + depends_on: + - zookeeper + - broker + ports: + - "8081:8081" + environment: + SCHEMA_REGISTRY_HOST_NAME: schema-registry + SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + + connect: + image: cnfldemos/cp-server-connect-datagen:0.3.2-5.5.0 + hostname: connect + container_name: connect + depends_on: + - zookeeper + - broker + - schema-registry + ports: + - "8083:8083" + environment: + CONNECT_BOOTSTRAP_SERVERS: 'broker:29092' + CONNECT_REST_ADVERTISED_HOST_NAME: connect + CONNECT_REST_PORT: 8083 + CONNECT_GROUP_ID: compose-connect-group + CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs + CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1 + CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000 + CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets + CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1 + CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status + CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1 + CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter + CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter + CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081 + CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter" + CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter" + CONNECT_ZOOKEEPER_CONNECT: 'zookeeper:2181' + # CLASSPATH required due to CC-2422 + CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-5.5.0.jar + CONNECT_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor" + CONNECT_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor" + CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components" + CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR + + control-center: + image: confluentinc/cp-enterprise-control-center:5.5.0 + hostname: control-center + container_name: control-center + depends_on: + - zookeeper + - broker + - schema-registry + - connect + - ksqldb-server + ports: + - "9021:9021" + environment: + CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092' + CONTROL_CENTER_ZOOKEEPER_CONNECT: 'zookeeper:2181' + CONTROL_CENTER_CONNECT_CLUSTER: 'connect:8083' + CONTROL_CENTER_KSQL_KSQLDB1_URL: "http://ksqldb-server:8088" + CONTROL_CENTER_KSQL_KSQLDB1_ADVERTISED_URL: "http://localhost:8088" + CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081" + CONTROL_CENTER_REPLICATION_FACTOR: 1 + CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1 + CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1 + CONFLUENT_METRICS_TOPIC_REPLICATION: 1 + PORT: 9021 + + ksqldb-server: + image: confluentinc/cp-ksqldb-server:5.5.0 + hostname: ksqldb-server + container_name: ksqldb-server + depends_on: + - broker + - connect + ports: + - "8088:8088" + environment: + KSQL_CONFIG_DIR: "/etc/ksql" + KSQL_BOOTSTRAP_SERVERS: "broker:29092" + KSQL_HOST_NAME: ksqldb-server + KSQL_LISTENERS: "http://0.0.0.0:8088" + KSQL_CACHE_MAX_BYTES_BUFFERING: 0 + KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081" + KSQL_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor" + KSQL_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor" + KSQL_KSQL_CONNECT_URL: "http://connect:8083" + + ksqldb-cli: + image: confluentinc/cp-ksqldb-cli:5.5.0 + container_name: ksqldb-cli + depends_on: + - broker + - connect + - ksqldb-server + entrypoint: /bin/sh + tty: true + + ksql-datagen: + image: confluentinc/ksqldb-examples:5.5.0 + hostname: ksql-datagen + container_name: ksql-datagen + depends_on: + - ksqldb-server + - broker + - schema-registry + - connect + command: "bash -c 'echo Waiting for Kafka to be ready... && \ + cub kafka-ready -b broker:29092 1 40 && \ + echo Waiting for Confluent Schema Registry to be ready... && \ + cub sr-ready schema-registry 8081 40 && \ + echo Waiting a few seconds for topic creation to finish... && \ + sleep 11 && \ + tail -f /dev/null'" + environment: + KSQL_CONFIG_DIR: "/etc/ksql" + STREAMS_BOOTSTRAP_SERVERS: broker:29092 + STREAMS_SCHEMA_REGISTRY_HOST: schema-registry + STREAMS_SCHEMA_REGISTRY_PORT: 8081 + + rest-proxy: + image: confluentinc/cp-kafka-rest:5.5.0 + depends_on: + - zookeeper + - broker + - schema-registry + ports: + - 8082:8082 + hostname: rest-proxy + container_name: rest-proxy + environment: + KAFKA_REST_HOST_NAME: rest-proxy + KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:29092' + KAFKA_REST_LISTENERS: "http://0.0.0.0:8082" + KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081' diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/pom.xml b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/pom.xml index e864ebe..7ff4c04 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/pom.xml +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/pom.xml @@ -11,14 +11,14 @@ org.springframework.boot spring-boot-starter-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.7.RELEASE 1.8.2 - 4.0.0 - Hoxton.BUILD-SNAPSHOT + 5.2.0 + Hoxton.SR5 @@ -38,6 +38,11 @@ org.springframework.cloud spring-cloud-stream-binder-kafka + + org.apache.avro + avro + ${avro.version} + io.confluent kafka-avro-serializer @@ -148,4 +153,4 @@ - \ No newline at end of file + diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/java/sample/consumer/ConfluentAvroConsumerApplication.java b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/java/sample/consumer/ConfluentAvroConsumerApplication.java index bee0aab..732ee2b 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/java/sample/consumer/ConfluentAvroConsumerApplication.java +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/java/sample/consumer/ConfluentAvroConsumerApplication.java @@ -7,9 +7,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.messaging.Sink; import org.springframework.context.annotation.Bean; @SpringBootApplication @@ -26,4 +23,4 @@ public class ConfluentAvroConsumerApplication { return input -> logger.info("input: " + input); } -} \ No newline at end of file +} diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/resources/application.yml b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/resources/application.yml index 37b9e66..342faeb 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/resources/application.yml +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/resources/application.yml @@ -2,16 +2,25 @@ spring: cloud: stream: bindings: - input: + process-in-0: destination: sensor-topic consumer: useNativeDecoding: true kafka: + +# binder: +# consumerProperties: +# value: +# subject: +# name: +# strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy + bindings: - input: + process-in-0: consumer: configuration: value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer schema.registry.url: http://localhost:8081 specific.avro.reader: true -server.port: 9999 \ No newline at end of file + +server.port: 9999 diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/pom.xml b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/pom.xml index 8cf4d41..0de96d9 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/pom.xml +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/pom.xml @@ -11,14 +11,14 @@ org.springframework.boot spring-boot-starter-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.7.RELEASE 1.8.2 - 4.0.0 - Hoxton.BUILD-SNAPSHOT + 5.2.0 + Hoxton.SR5 @@ -35,6 +35,19 @@ + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter-web + + + org.apache.avro + avro + ${avro.version} + org.springframework.cloud spring-cloud-stream-binder-kafka @@ -161,4 +174,4 @@ - \ No newline at end of file + diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/src/main/resources/application.yml b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/src/main/resources/application.yml index 0ba468d..79a3702 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/src/main/resources/application.yml +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/src/main/resources/application.yml @@ -2,15 +2,23 @@ spring: cloud: stream: bindings: - output: + supplier-out-0: destination: sensor-topic producer: useNativeEncoding: true - kafka: + + kafka: +# binder: +# producerProperties: +# value: +# subject: +# name: +# strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy + bindings: - output: + supplier-out-0: producer: configuration: value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer schema.registry.url: http://localhost:8081 -server.port: 9009 \ No newline at end of file +server.port: 9009 diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/pom.xml b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/pom.xml index 5c57d5e..e88aab7 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/pom.xml +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/pom.xml @@ -11,14 +11,14 @@ org.springframework.boot spring-boot-starter-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.7.RELEASE 1.8.2 - 4.0.0 - Hoxton.BUILD-SNAPSHOT + 5.2.0 + Hoxton.SR5 @@ -34,6 +34,19 @@ + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter-web + + + org.apache.avro + avro + ${avro.version} + org.springframework.cloud spring-cloud-stream-binder-kafka @@ -160,4 +173,4 @@ - \ No newline at end of file + diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/src/main/resources/application.yml b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/src/main/resources/application.yml index 7b70ce3..1a14d9c 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/src/main/resources/application.yml +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/src/main/resources/application.yml @@ -7,10 +7,17 @@ spring: producer: useNativeEncoding: true kafka: + # binder: + # producerProperties: + # value: + # subject: + # name: + # strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy + bindings: output: producer: configuration: value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer schema.registry.url: http://localhost:8081 -server.port: 9010 \ No newline at end of file +server.port: 9010 diff --git a/schema-registry-samples/schema-registry-confluent/README.adoc b/schema-registry-samples/schema-registry-confluent/README.adoc index 6b70c44..9225988 100644 --- a/schema-registry-samples/schema-registry-confluent/README.adoc +++ b/schema-registry-samples/schema-registry-confluent/README.adoc @@ -110,4 +110,57 @@ Once you are done with running the samples, stop the docker containers `docker-compose down` -and stop the Confluent Schema Registry server. \ No newline at end of file +and stop the Confluent Schema Registry server. + +==== Using Confluent Command Center + +This test can be run with the [Confluent Command Center](https://docs.confluent.io/current/control-center/index.html) - a web-based tool for managing and monitoring Apache Kafka®. Control Center facilitates building and monitoring production data pipelines and streaming applications. + +For a quick start you can install the command center with the help of the `docker-compose-control-center.yaml`: + +``` +docker-compose -f ./docker-compose-control-center.yaml +``` + +For further info check the Confluent's [Quick Start](https://docs.confluent.io/current/quickstart/ce-docker-quickstart.html). +The [Schema Registry feature in Control Center](https://docs.confluent.io/current/control-center/topics/schema.html) would help you to visualize and manage the topic schemas. + +After you run the samples and post couple of messages as explained above. + +1. Open the command center at `http://localhost:9021` and click on the provided cluster. +2. From the vertical menu select `Topics` tab. +3. From the list of topics select the `sensor-topic` - the topic created by the samples. +4. Click on the `Schema` tab to see the `Sensors` schema. + +You can also use the Confluent Schema REST API at `http://localhost:8081`. For example the `http://localhost:8081/subjects` will list the schema names (e.g. subjects) defined. +After you have run the samples you should be able to see a schema subject name `sensor-topic-value`. + +===== NOTE + +Note that currently the Command Center recognizes only schema subjects created with [TopicNameStrategy](https://docs.confluent.io/current/schema-registry/serdes-develop/index.html). +That means that the schema name (e.g. subject) must be named after your topic name (e.g. spring.cloud.stream.bindings.:destination) followed by `-value` suffix. +In our sample the topic (e.g. destination) is called `sensor-topic`. Therefore the schema subject should be named `sensor-topic-value` to be recognizable by the Confluent command center. + +Note that the confluent schema registry works fine with different subject naming strategies, only the command center visualizaiton will not show them. + +To configure a the schema name after the topic name yuo can use the `spring.cloud.schema.avro.subjectNamePrefix` +and `spring.cloud.schema.avro.subjectNamePrefix.subjectNamingStrategy` properties as shown below (only available with 1.1.x onwards: + +``` +spring: + cloud: + stream: + bindings: + process-in-0: + destination: sensor-topic + schemaRegistryClient: + endpoint: http://localhost:8081 + schema: + avro: + schema-locations: classpath:avro/sensor.avsc + subjectNamePrefix: sensor-topic-value + subjectNamingStrategy: org.springframework.cloud.schema.registry.avro.SubjectPrefixOnlyNamingStrategy +server.port: 9999 +``` + +The `subjectNamePrefix` must be set to the value of your `destination` followed by `-value` suffix. The `subjectNamingStrategy` must be set to `SubjectPrefixOnlyNamingStrategy`. diff --git a/schema-registry-samples/schema-registry-confluent/docker-compose-control-center.yaml b/schema-registry-samples/schema-registry-confluent/docker-compose-control-center.yaml new file mode 100644 index 0000000..85ed076 --- /dev/null +++ b/schema-registry-samples/schema-registry-confluent/docker-compose-control-center.yaml @@ -0,0 +1,178 @@ +--- +version: '2' +services: + zookeeper: + image: confluentinc/cp-zookeeper:5.5.0 + hostname: zookeeper + container_name: zookeeper + ports: + - "2181:2181" + environment: + ZOOKEEPER_CLIENT_PORT: 2181 + ZOOKEEPER_TICK_TIME: 2000 + + broker: + image: confluentinc/cp-server:5.5.0 + hostname: broker + container_name: broker + depends_on: + - zookeeper + ports: + - "9092:9092" + environment: + KAFKA_BROKER_ID: 1 + KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181' + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092 + KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 + KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 + KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 + CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092 + CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper:2181 + CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1 + CONFLUENT_METRICS_ENABLE: 'true' + CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous' + + schema-registry: + image: confluentinc/cp-schema-registry:5.5.0 + hostname: schema-registry + container_name: schema-registry + depends_on: + - zookeeper + - broker + ports: + - "8081:8081" + environment: + SCHEMA_REGISTRY_HOST_NAME: schema-registry + SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + + connect: + image: cnfldemos/cp-server-connect-datagen:0.3.2-5.5.0 + hostname: connect + container_name: connect + depends_on: + - zookeeper + - broker + - schema-registry + ports: + - "8083:8083" + environment: + CONNECT_BOOTSTRAP_SERVERS: 'broker:29092' + CONNECT_REST_ADVERTISED_HOST_NAME: connect + CONNECT_REST_PORT: 8083 + CONNECT_GROUP_ID: compose-connect-group + CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs + CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1 + CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000 + CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets + CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1 + CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status + CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1 + CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter + CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter + CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081 + CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter" + CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter" + CONNECT_ZOOKEEPER_CONNECT: 'zookeeper:2181' + # CLASSPATH required due to CC-2422 + CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-5.5.0.jar + CONNECT_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor" + CONNECT_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor" + CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components" + CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR + + control-center: + image: confluentinc/cp-enterprise-control-center:5.5.0 + hostname: control-center + container_name: control-center + depends_on: + - zookeeper + - broker + - schema-registry + - connect + - ksqldb-server + ports: + - "9021:9021" + environment: + CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092' + CONTROL_CENTER_ZOOKEEPER_CONNECT: 'zookeeper:2181' + CONTROL_CENTER_CONNECT_CLUSTER: 'connect:8083' + CONTROL_CENTER_KSQL_KSQLDB1_URL: "http://ksqldb-server:8088" + CONTROL_CENTER_KSQL_KSQLDB1_ADVERTISED_URL: "http://localhost:8088" + CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081" + CONTROL_CENTER_REPLICATION_FACTOR: 1 + CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1 + CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1 + CONFLUENT_METRICS_TOPIC_REPLICATION: 1 + PORT: 9021 + + ksqldb-server: + image: confluentinc/cp-ksqldb-server:5.5.0 + hostname: ksqldb-server + container_name: ksqldb-server + depends_on: + - broker + - connect + ports: + - "8088:8088" + environment: + KSQL_CONFIG_DIR: "/etc/ksql" + KSQL_BOOTSTRAP_SERVERS: "broker:29092" + KSQL_HOST_NAME: ksqldb-server + KSQL_LISTENERS: "http://0.0.0.0:8088" + KSQL_CACHE_MAX_BYTES_BUFFERING: 0 + KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081" + KSQL_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor" + KSQL_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor" + KSQL_KSQL_CONNECT_URL: "http://connect:8083" + + ksqldb-cli: + image: confluentinc/cp-ksqldb-cli:5.5.0 + container_name: ksqldb-cli + depends_on: + - broker + - connect + - ksqldb-server + entrypoint: /bin/sh + tty: true + + ksql-datagen: + image: confluentinc/ksqldb-examples:5.5.0 + hostname: ksql-datagen + container_name: ksql-datagen + depends_on: + - ksqldb-server + - broker + - schema-registry + - connect + command: "bash -c 'echo Waiting for Kafka to be ready... && \ + cub kafka-ready -b broker:29092 1 40 && \ + echo Waiting for Confluent Schema Registry to be ready... && \ + cub sr-ready schema-registry 8081 40 && \ + echo Waiting a few seconds for topic creation to finish... && \ + sleep 11 && \ + tail -f /dev/null'" + environment: + KSQL_CONFIG_DIR: "/etc/ksql" + STREAMS_BOOTSTRAP_SERVERS: broker:29092 + STREAMS_SCHEMA_REGISTRY_HOST: schema-registry + STREAMS_SCHEMA_REGISTRY_PORT: 8081 + + rest-proxy: + image: confluentinc/cp-kafka-rest:5.5.0 + depends_on: + - zookeeper + - broker + - schema-registry + ports: + - 8082:8082 + hostname: rest-proxy + container_name: rest-proxy + environment: + KAFKA_REST_HOST_NAME: rest-proxy + KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:29092' + KAFKA_REST_LISTENERS: "http://0.0.0.0:8082" + KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081' diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/pom.xml b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/pom.xml index 219afc2..3cf5a2e 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/pom.xml +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/pom.xml @@ -11,14 +11,14 @@ org.springframework.boot spring-boot-starter-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.7.RELEASE - 1.8.2 - 4.0.0 - Hoxton.BUILD-SNAPSHOT + 1.9.2 + 5.2.0 + Hoxton.SR5 @@ -35,6 +35,19 @@ + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter-web + + + org.apache.avro + avro + ${avro.version} + org.springframework.cloud spring-cloud-stream-binder-kafka @@ -42,6 +55,7 @@ org.springframework.cloud spring-cloud-schema-registry-client + 1.1.0.BUILD-SNAPSHOT org.apache.avro @@ -160,4 +174,4 @@ - \ No newline at end of file + diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/src/main/resources/application.yml b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/src/main/resources/application.yml index 9587c5c..2dfa597 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/src/main/resources/application.yml +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/src/main/resources/application.yml @@ -6,7 +6,9 @@ spring: destination: sensor-topic schemaRegistryClient: endpoint: http://localhost:8081 - schema: - avro: - schema-locations: classpath:avro/sensor.avsc -server.port: 9999 \ No newline at end of file + schema: + avro: + schema-locations: classpath:avro/sensor.avsc + subjectNamePrefix: sensor-topic-value + subjectNamingStrategy: org.springframework.cloud.schema.registry.avro.SubjectPrefixOnlyNamingStrategy +server.port: 9999 diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/pom.xml b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/pom.xml index bb541c2..87f32b9 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/pom.xml +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/pom.xml @@ -11,14 +11,14 @@ org.springframework.boot spring-boot-starter-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.7.RELEASE - 1.8.2 - 4.0.0 - Hoxton.BUILD-SNAPSHOT + 1.9.2 + 5.2.0 + Hoxton.SR5 @@ -34,6 +34,19 @@ + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter-web + + + org.apache.avro + avro + ${avro.version} + org.springframework.cloud spring-cloud-stream-binder-kafka @@ -41,6 +54,7 @@ org.springframework.cloud spring-cloud-schema-registry-client + 1.1.0.BUILD-SNAPSHOT org.apache.avro @@ -158,4 +172,4 @@ - \ No newline at end of file + diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/src/main/resources/application.yml b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/src/main/resources/application.yml index ca58ea1..62f8fd1 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/src/main/resources/application.yml +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/src/main/resources/application.yml @@ -7,7 +7,10 @@ spring: destination: sensor-topic schemaRegistryClient: endpoint: http://localhost:8081 - schema: - avro: - schema-locations: classpath:avro/sensor.avsc -server.port: 9009 \ No newline at end of file + schema: + avro: + schema-locations: classpath:avro/sensor.avsc + subjectNamePrefix: sensor-topic-value + subjectNamingStrategy: org.springframework.cloud.schema.registry.avro.SubjectPrefixOnlyNamingStrategy + +server.port: 9009 diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/pom.xml b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/pom.xml index 22cfb26..9c08f57 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/pom.xml +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/pom.xml @@ -11,14 +11,14 @@ org.springframework.boot spring-boot-starter-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.7.RELEASE - 1.8.2 - 4.0.0 - Hoxton.BUILD-SNAPSHOT + 1.9.2 + 5.2.0 + Hoxton.SR5 @@ -34,6 +34,19 @@ + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter-web + + + org.apache.avro + avro + ${avro.version} + org.springframework.cloud spring-cloud-stream-binder-kafka @@ -158,4 +171,4 @@ - \ No newline at end of file +