Confluent-Kafka-Springboot

A comprehensive, production-ready example of integrating Spring Boot 3.x with Confluent Cloud. This repository demonstrates how to manage multiple Kafka topics using different serialization formats (standard JSON/String and Avro), secure connections using SASL_SSL, and handle robust dependency management.

Overview

This project serves as a robust template for developing microservices that interact with Confluent Cloud. It tackles common challenges developers face when working with a multi-protocol (JSON/Avro) environment, providing clear, tested configurations for producers, consumers, and even Kafka Streams.

Key Features:

• Dual Serialization: Configuration for both standard String/Integer serialization and Avro serialization via the Confluent Schema Registry.
• Secure Connectivity: Uses SASL_SSL (PLAIN mechanism) for authentication with Confluent Cloud Kafka brokers and the Schema Registry.
• Modern Spring Boot: Built with Spring Boot 3.5.x and Java 17+, leveraging Jakarta EE APIs.
• Reactive Producers: Uses Project Reactor (Flux) for non-blocking message generation at startup.
• Lombok & DI: Utilizes @RequiredArgsConstructor for clean dependency injection and @Qualifier for managing multiple Kafka template beans.

---

🛠️ Prerequisites

To run this application, you will need:

1. Java 17+ installed.
2. Gradle (managed by the wrapper ./gradlew).
3. A Confluent Cloud Account with a Kafka Cluster and a Schema Registry instance running.
4. API Keys for both your Kafka Cluster and your Schema Registry.

---

⚙️ Configuration

All configuration is managed within the KafkaConfig.java class using hardcoded values for clarity in this example.

1. Clone the repository: bash

   git clone github.com
   cd Confluent-Kafka-Springboot
   

   Use code with caution.

2. **Update src/main/java/com/hibersoft/springkafka/KafkaConfig.java:**Replace the placeholder values with your actual Confluent Cloud credentials: java

   // --- HARDCODED CONFIGURATION VALUES ---
   private final String BOOTSTRAP_SERVERS = "pkc-xxxx.us-east-1.gcp.confluent.cloud:9092";
   private final String API_KEY = "YOUR_KAFKA_CLUSTER_API_KEY"; // Cluster Key
   private final String API_SECRET = "YOUR_KAFKA_CLUSTER_API_SECRET"; // Cluster Secret
   // ...
   private final String SCHEMA_REGISTRY_URL = "https://psrc-xxxx.us-east-1.confluent.cloud";
   private final String SCHEMA_REGISTRY_API_KEY = "YOUR_SR_API_KEY"; // Schema Registry Key
   private final String SCHEMA_REGISTRY_API_SECRET = "YOUR_SR_API_SECRET"; // Schema Registry Secret
   // ...
   

---

🚀 Running the Application

Use the Gradle bootRun task to compile, build, and run the Spring Boot application:

bash

./gradlew bootRun

Use code with caution.

Expected Behavior:

1. Startup: The application will quiet down verbose Spring logs (configured via application.properties) and print application-specific logs.
2. Production: The Producer service uses reactive streams (Flux.interval) to asynchronously send messages every 10 seconds to two topics:
   • hobbit (String/Integer serialization)
   • hobbit-avro (Avro serialization)
3. Consumption: The Consumer service is running listeners configured to handle the specific schemas for both topics correctly.
4. Logs: You will see continuous logs confirming messages being produced and consumed.

---

📂 Project Structure Highlights

File/Class	Purpose	Key Annotations/Concepts
KafkaConfig.java	Centralized bean definitions for all Kafka components.	@Configuration, @Bean, @Qualifier
Producer.java	Handles outbound messages. Uses reactive streams for continuous production.	@Service, @RequiredArgsConstructor, @EventListener
Consumer.java	Handles inbound messages from topics.	@Service, @KafkaListener, @Qualifier
build.gradle	Manages dependencies, including crucial Confluent 7.5.0 compatibility.	io.confluent:kafka-avro-serializer:7.5.0
application.properties	Manages core Spring behavior and logging levels (logging.level.root=WARN, spring.main.lazy-initialization=false).	Properties management

---

💡 Key Takeaways from Development

This project was built iteratively, resolving several complex issues encountered in the wild:

• Dependency Hell: A major challenge was aligning older Confluent 6.x libraries (using javax APIs) with modern Spring Boot 3.x (using jakarta APIs). The fix was upgrading Confluent libraries to version 7.5.0+.
• Authentication 401s: We resolved connection errors by ensuring dedicated API keys were used for the Kafka Cluster AND the Schema Registry separately.
• Debugging Silent Errors: We found that silent failures in configuration required forcing eager bean initialization (spring.main.lazy-initialization=false) and reducing overall logging verbosity (logging.level.root=WARN).

---

Contributing

Feel free to open issues, submit PRs, or fork this repository for your own learning purposes.