Added message sending (#85)

* Added message sending

* sent message to kafka

* renamed docker service

* renamed message file

* Changed message to json

* fixed docker compose and messages

Co-authored-by: Roman Nedzvetskiy <roman@Romans-MacBook-Pro.local>
Co-authored-by: German Osin <german.osin@gmail.com>

docker/kafka-clusters-only.yaml

@@ -5,113 +5,67 @@ services:
   zookeeper0:
     image: confluentinc/cp-zookeeper:5.1.0
     environment:
-      ZOOKEEPER_CLIENT_PORT: 2183
+      ZOOKEEPER_CLIENT_PORT: 2181
       ZOOKEEPER_TICK_TIME: 2000
-    ports:
-     - 2183:2183
 
   kafka0:
     image: confluentinc/cp-kafka:5.1.0
     depends_on:
       - zookeeper0
-    ports:
-      - 29091:29091
-      - 9997:9997
     environment:
       KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper0:2183
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka0:9092,PLAINTEXT_HOST://localhost:29091 #,PLAIN://kafka0:29090
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT #,PLAIN:PLAINTEXT
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper0:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka0:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
-      JMX_PORT: 9997
-      KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Dcom.sun.management.jmxremote.rmi.port=9997
-
-  kafka01:
-    image: confluentinc/cp-kafka:5.1.0
-    depends_on:
-      - zookeeper0
-    ports:
-      - 29093:29093
-      - 9999:9999
-    environment:
-      KAFKA_BROKER_ID: 2
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper0:2183
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka01:9092,PLAINTEXT_HOST://localhost:29093,PLAIN://kafka0:29090
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,PLAIN:PLAINTEXT
-      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       JMX_PORT: 9997
-      KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Dcom.sun.management.jmxremote.rmi.port=9997
-
-  kafka-init-topics0:
-    image: confluentinc/cp-kafka:5.1.0
-    depends_on:
-      - kafka0
-    command: "bash -c 'echo Waiting for Kafka to be ready... && \
-                cub kafka-ready -b kafka0:9092 1 20 && \
-                kafka-topics --create --topic users --partitions 2 --replication-factor 2 --if-not-exists --zookeeper zookeeper0:2183 && \
-                kafka-topics --create --topic messages --partitions 3 --replication-factor 2 --if-not-exists --zookeeper zookeeper0:2183'"
-    environment:
-      KAFKA_BROKER_ID: ignored
-      KAFKA_ZOOKEEPER_CONNECT: ignored
-    networks:
-      - default
+      KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=kafka0 -Dcom.sun.management.jmxremote.rmi.port=9997
 
   zookeeper1:
     image: confluentinc/cp-zookeeper:5.1.0
     environment:
-      ZOOKEEPER_CLIENT_PORT: 2182
+      ZOOKEEPER_CLIENT_PORT: 2181
       ZOOKEEPER_TICK_TIME: 2000
-    ports:
-      - 2182:2182
 
   kafka1:
     image: confluentinc/cp-kafka:5.1.0
     depends_on:
       - zookeeper1
-    ports:
-      - 29092:29092
-      - 9998:9998
     environment:
       KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper1:2182
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,PLAINTEXT_HOST://localhost:29092,PLAIN://localhost:29090
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,PLAIN:PLAINTEXT
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper1:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:29092,PLAINTEXT_HOST://localhost:9093
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      JMX_PORT: 9998
-      KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Dcom.sun.management.jmxremote.rmi.port=9998
-
-  kafka-init-topics1:
-    image: confluentinc/cp-kafka:5.1.0
-    depends_on:
-     - kafka1
-    command: "bash -c 'echo Waiting for Kafka to be ready... && \
-               cub kafka-ready -b kafka1:9092 20 && \
-               kafka-topics --create --topic secondUsers --partitions 3 --replication-factor 2 --if-not-exists --zookeeper zookeeper1:2182 && \
-               kafka-topics --create --topic secondMessages --partitions 2 --replication-factor 2 --if-not-exists --zookeeper zookeeper1:2182'"
-    environment:
-     KAFKA_BROKER_ID: ignored
-     KAFKA_ZOOKEEPER_CONNECT: ignored
-    networks:
-     - default
+      JMX_PORT: 9997
+      KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=kafka1 -Dcom.sun.management.jmxremote.rmi.port=9997
 
   schemaregistry0:
     image: confluentinc/cp-schema-registry:5.1.0
     depends_on:
       - zookeeper0
       - kafka0
-      - kafka01
-    ports:
-      - 8085:8085
     environment:
-      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka0:9092,PLAINTEXT://kafka01:9092
-      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper0:2183
+      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka0:29092
+      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper0:2181
       SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL: PLAINTEXT
-      SCHEMA_REGISTRY_HOST_NAME: schemaregistry
-      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8085
+      SCHEMA_REGISTRY_HOST_NAME: schemaregistry0
+      SCHEMA_REGISTRY_LISTENERS: http://schemaregistry0:8085
 
       SCHEMA_REGISTRY_SCHEMA_REGISTRY_INTER_INSTANCE_PROTOCOL: "http"
       SCHEMA_REGISTRY_LOG4J_ROOT_LOGLEVEL: INFO
       SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas
+
+  kafka-init-topics:
+    image: confluentinc/cp-kafka:5.1.0
+    volumes:
+       - ./message.json:/data/message.json
+    depends_on:
+      - kafka1
+    command: "bash -c 'echo Waiting for Kafka to be ready... && \
+               cub kafka-ready -b kafka1:29092 1 30 && \
+               kafka-topics --create --topic second.users --partitions 3 --replication-factor 1 --if-not-exists --zookeeper zookeeper1:2181 && \
+               kafka-topics --create --topic second.messages --partitions 2 --replication-factor 1 --if-not-exists --zookeeper zookeeper1:2181 && \
+               kafka-console-producer --broker-list kafka1:29092 -topic second.users < /data/message.json'"

docker/kafka-ui.yaml

@@ -13,7 +13,14 @@ services:
       - kafka0
       - kafka1
       - schemaregistry0
-    command: [ "java", "-jar", "kafka-ui-api.jar", "--spring.profiles.active=sdp"]
+    environment:
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka0:29092
+      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper0:2181
+      KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schemaregistry0:8085
+      KAFKA_CLUSTERS_1_NAME: secondLocal
+      KAFKA_CLUSTERS_1_BOOTSTRAPSERVERS: kafka1:29092
+      KAFKA_CLUSTERS_1_ZOOKEEPER: zookeeper1:2181
 
   zookeeper0:
     image: confluentinc/cp-zookeeper:5.1.0
@@ -48,7 +55,7 @@ services:
     environment:
       KAFKA_BROKER_ID: 1
       KAFKA_ZOOKEEPER_CONNECT: zookeeper1:2181
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:29092,PLAINTEXT_HOST://localhost:9093
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
@@ -70,3 +77,15 @@ services:
       SCHEMA_REGISTRY_SCHEMA_REGISTRY_INTER_INSTANCE_PROTOCOL: "http"
       SCHEMA_REGISTRY_LOG4J_ROOT_LOGLEVEL: INFO
       SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas
+
+  kafka-init-topics:
+    image: confluentinc/cp-kafka:5.1.0
+    volumes:
+       - ./message.json:/data/message.json
+    depends_on:
+      - kafka1
+    command: "bash -c 'echo Waiting for Kafka to be ready... && \
+               cub kafka-ready -b kafka1:29092 1 30 && \
+               kafka-topics --create --topic second.users --partitions 3 --replication-factor 1 --if-not-exists --zookeeper zookeeper1:2181 && \
+               kafka-topics --create --topic second.messages --partitions 2 --replication-factor 1 --if-not-exists --zookeeper zookeeper1:2181 && \
+               kafka-console-producer --broker-list kafka1:29092 -topic second.users < /data/message.json'"

docker/message.json

@@ -0,0 +1,100 @@
+{"message":"hello1","destination":"kafka2"}
+{"message":"hello2","destination":"kafka2"}
+{"message":"hello3","destination":"kafka2"}
+{"message":"hello4","destination":"kafka2"}
+{"message":"hello5","destination":"kafka2"}
+{"message":"hello6","destination":"kafka2"}
+{"message":"hello7","destination":"kafka2"}
+{"message":"hello8","destination":"kafka2"}
+{"message":"hello9","destination":"kafka2"}
+{"message":"hello10","destination":"kafka2"}
+{"message":"hello11","destination":"kafka2"}
+{"message":"hello12","destination":"kafka2"}
+{"message":"hello13","destination":"kafka2"}
+{"message":"hello14","destination":"kafka2"}
+{"message":"hello15","destination":"kafka2"}
+{"message":"hello16","destination":"kafka2"}
+{"message":"hello17","destination":"kafka2"}
+{"message":"hello18","destination":"kafka2"}
+{"message":"hello19","destination":"kafka2"}
+{"message":"hello20","destination":"kafka2"}
+{"message":"hello21","destination":"kafka2"}
+{"message":"hello22","destination":"kafka2"}
+{"message":"hello23","destination":"kafka2"}
+{"message":"hello24","destination":"kafka2"}
+{"message":"hello25","destination":"kafka2"}
+{"message":"hello26","destination":"kafka2"}
+{"message":"hello27","destination":"kafka2"}
+{"message":"hello28","destination":"kafka2"}
+{"message":"hello29","destination":"kafka2"}
+{"message":"hello30","destination":"kafka2"}
+{"message":"hello31","destination":"kafka2"}
+{"message":"hello32","destination":"kafka2"}
+{"message":"hello33","destination":"kafka2"}
+{"message":"hello34","destination":"kafka2"}
+{"message":"hello35","destination":"kafka2"}
+{"message":"hello36","destination":"kafka2"}
+{"message":"hello37","destination":"kafka2"}
+{"message":"hello38","destination":"kafka2"}
+{"message":"hello39","destination":"kafka2"}
+{"message":"hello40","destination":"kafka2"}
+{"message":"hello41","destination":"kafka2"}
+{"message":"hello42","destination":"kafka2"}
+{"message":"hello43","destination":"kafka2"}
+{"message":"hello44","destination":"kafka2"}
+{"message":"hello45","destination":"kafka2"}
+{"message":"hello46","destination":"kafka2"}
+{"message":"hello47","destination":"kafka2"}
+{"message":"hello48","destination":"kafka2"}
+{"message":"hello49","destination":"kafka2"}
+{"message":"hello50","destination":"kafka2"}
+{"message":"hello51","destination":"kafka2"}
+{"message":"hello52","destination":"kafka2"}
+{"message":"hello53","destination":"kafka2"}
+{"message":"hello54","destination":"kafka2"}
+{"message":"hello55","destination":"kafka2"}
+{"message":"hello56","destination":"kafka2"}
+{"message":"hello57","destination":"kafka2"}
+{"message":"hello58","destination":"kafka2"}
+{"message":"hello59","destination":"kafka2"}
+{"message":"hello60","destination":"kafka2"}
+{"message":"hello61","destination":"kafka2"}
+{"message":"hello62","destination":"kafka2"}
+{"message":"hello63","destination":"kafka2"}
+{"message":"hello64","destination":"kafka2"}
+{"message":"hello65","destination":"kafka2"}
+{"message":"hello66","destination":"kafka2"}
+{"message":"hello67","destination":"kafka2"}
+{"message":"hello68","destination":"kafka2"}
+{"message":"hello69","destination":"kafka2"}
+{"message":"hello70","destination":"kafka2"}
+{"message":"hello71","destination":"kafka2"}
+{"message":"hello72","destination":"kafka2"}
+{"message":"hello73","destination":"kafka2"}
+{"message":"hello74","destination":"kafka2"}
+{"message":"hello75","destination":"kafka2"}
+{"message":"hello76","destination":"kafka2"}
+{"message":"hello77","destination":"kafka2"}
+{"message":"hello78","destination":"kafka2"}
+{"message":"hello79","destination":"kafka2"}
+{"message":"hello80","destination":"kafka2"}
+{"message":"hello81","destination":"kafka2"}
+{"message":"hello82","destination":"kafka2"}
+{"message":"hello83","destination":"kafka2"}
+{"message":"hello84","destination":"kafka2"}
+{"message":"hello85","destination":"kafka2"}
+{"message":"hello86","destination":"kafka2"}
+{"message":"hello87","destination":"kafka2"}
+{"message":"hello88","destination":"kafka2"}
+{"message":"hello89","destination":"kafka2"}
+{"message":"hello90","destination":"kafka2"}
+{"message":"hello91","destination":"kafka2"}
+{"message":"hello92","destination":"kafka2"}
+{"message":"hello93","destination":"kafka2"}
+{"message":"hello94","destination":"kafka2"}
+{"message":"hello95","destination":"kafka2"}
+{"message":"hello96","destination":"kafka2"}
+{"message":"hello97","destination":"kafka2"}
+{"message":"hello98","destination":"kafka2"}
+{"message":"hello99","destination":"kafka2"}
+{"message":"hello100","destination":"kafka2"}