Apache Kafka Proof of Concept - Integration with Spring Boot
- Confluent Apache Kafka
- Spring Boot
- MongoDB
- Docker
- TestContainers
Management system of a virtual sushi booth in which each customer can consult the online menu; once registered can place the order, indicate the table to be served and pay the bill. Once the bill has been paid, the cook appears with the order to cook and serve to the customer his dinner / lunch. A room administrator has a dashboard available to manage users, customize and compose the menu with the prices of the courses, supervise the work of the cook.
- java version 21
- maven 3.9.3
- docker 25.0.3 (https://www.docker.com/)
Based on Apache Kafka & spring boot.
- app: main application
- cook: process the order(s)
Before starting the app, you need to register oauth2 on google cloud console.
This will generate the client Id and the Client Secret. Then replace the default ones in the application.yml
security:
oauth2:
client:
registration:
google :
clientId: <YOUR-CLIENT-ID>
clientSecret: <YOUR-CLIENT-SECRET>
From the scripts folder.
$ docker compose up -d
$ docker compose down
$ docker exec -it sushibar-backend-app-kafka-1 bash (For Windows add winpty before docker)
$ /bin/kafka-topics --list --bootstrap-server localhost:29092
$ /bin/kafka-console-consumer --bootstrap-server localhost:29092 --topic orders --from-beginning
/bin/kafka-topics --bootstrap-server localhost:9092 --delete --topic processed-orders
/bin/kafka-topics --bootstrap-server localhost:9092 --delete --topic orders
Connect to the docker kafka connector instance:
$ docker exec -it sushibar-backend-kafka_connect_1 bash
From the docker container install the kafka connector:
$ ./confluent-hub install --no-prompt mongodb/kafka-connect-mongodb:latest
Configure the kafka connector for the specific topic with postman.
- url: http://localhost:8083/connectors
- method: POST
- body:
{"name": "mongo-sink-connector",
"config": {
"connector.class":"com.mongodb.kafka.connect.MongoSinkConnector",
"connection.uri":"mongodb://root:rootpassword@mongodb_container",
"database":"order",
"collection":"sdp_order",
"topics":"orders",
"key.converter.schemas.enable":"true",
"value.converter.schemas.enable": "true",
"key.converter":"io.confluent.connect.avro.AvroConverter",
"key.converter.enhanced.avro.schema.support":true,
"key.converter.schema.registry.url": "http://schema-registry:8081",
"value.converter.schema.registry.url": "http://schema-registry:8081",
"value.converter.enhanced.avro.schema.support":true,
"value.converter": "io.confluent.connect.avro.AvroConverter"
}
}
http://localhost:8080/swagger-ui/index.html
org.apache.kafka.common.errors.InvalidConfigurationException: Schema being registered is incompatible with an earlier schema; error code: 409
Set the compatibility mode of the topic to NONE.