![image.png](https://prod-files-secure.s3.us-west-2.amazonaws.com/c26cbd8c-4ae9-4ab6-b595-99bb653f9d83/57d33b73-fa4b-4f6c-b25a-f7c2d827968b/image.png)

Q- First i have visited Spring Initilazer : and create this 

### **Initialize a Spring Boot Project**

You can use Spring Initializr or your IDE to create a new project.

- **Project**: Maven Project
- **Language**: Java
- **Spring Boot**: 2.3.9 (as per the example)
- **Project Metadata**:
    - Group: **`com.amir`**
    - Artifact: **`kafka-pub-sub`**
    - Name: **`kafka-pub-sub`**
    - Package Name: **`com.amir`**
    - Packaging: Jar
    - Java: 8 or higher
- **Dependencies**:
    - Spring Web
    - Spring for Apache Kafka
    - Lombok (optional)

Alternatively, you can manually create the project using your IDE.

Q- next let’s step up the project folder struture here 

### **Directory Structure**

Align your project structure as follows for clarity:

```
reactive-kafka-consumer-producer/
├── src/
│   ├── main/
│   │   ├── java/com/example/reactivekafkaconsumerandproducer/
│   │   │   ├── config/
│   │   │   ├── controller/
│   │   │   ├── dto/
│   │   │   ├── service/
│   │   │   ├── ReactiveKafkaConsumerProducerApplication.java
│   │   └── resources/
│   │       └── application.properties
│   └── test/
│       └── java/com/example/reactivekafkaconsumerandproducer/
└── pom.xml

```

### **Explanation**

- **config/**: Configuration classes for Kafka.
- **controller/**: REST controllers.
- **dto/**: Data Transfer Objects.
- **service/**: Services for producing and consuming messages.
- **application.properties**: Configuration properties.
- **tests/**: Integration and unit tests.

## **Define Data Transfer Objects (DTOs)**

Create DTOs representing the data you intend to send and consume via Kafka.

### **FakeProducerDTO.java**

```java
package com.example.reactivekafkaconsumerandproducer.dto;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class FakeProducerDTO {
    private String id;
    private String message;
}

```

### **FakeConsumerDTO.java**

```java
package com.example.reactivekafkaconsumerandproducer.dto;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class FakeConsumerDTO {
    private String id;
    private String content;
}

```

Q- next given this pom.xml file 

```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>3.3.5</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.amir</groupId>
	<artifactId>kafka-SpringBoot</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>kafka-SpringBoot</name>
	<description>Tutor to integrate kafka with Spring Boot </description>
	<url/>
	<licenses>
		<license/>
	</licenses>
	<developers>
		<developer/>
	</developers>
	<scm>
		<connection/>
		<developerConnection/>
		<tag/>
		<url/>
	</scm>
	<properties>
		<java.version>17</java.version>
		<reactor.kafka.version>1.3.21</reactor.kafka.version>
	</properties>
	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-devtools</artifactId>
			<scope>runtime</scope>
			<optional>true</optional>
		</dependency>
		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
			<optional>true</optional>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka-test</artifactId>
			<scope>test</scope>
		</dependency>
		 <!-- Reactor Kafka -->
        <dependency>
            <groupId>io.projectreactor.kafka</groupId>
            <artifactId>reactor-kafka</artifactId>
            <version>${reactor.kafka.version}</version>
        </dependency>

        <!-- JSON Serialization -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
				<configuration>
					<excludes>
						<exclude>
							<groupId>org.projectlombok</groupId>
							<artifactId>lombok</artifactId>
						</exclude>
					</excludes>
				</configuration>
			</plugin>
		</plugins>
	</build>

</project>

```

Q- we create the package called “Config” 

and this is the script

```java
package com.amir.kafka_SpringBoot.Config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import com.amir.kafka_SpringBoot.dto.FakeConsumerDTO;

import reactor.kafka.receiver.ReceiverOptions;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class ReactiveKafkaConsumerConfig {

    @Value("${FAKE_CONSUMER_DTO_TOPIC}")
    private String topic;

    @Bean
    public ReceiverOptions<String, FakeConsumerDTO> kafkaReceiverOptions(KafkaProperties kafkaProperties) {
        Map<String, Object> props = new HashMap<>(kafkaProperties.buildConsumerProperties());

        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, 
                 "com.amir.kafka_SpringBoot.dto.FakeConsumerDTO");
        props.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, false);

        ReceiverOptions<String, FakeConsumerDTO> basicReceiverOptions = ReceiverOptions.create(props);
        return basicReceiverOptions.subscription(Collections.singletonList(topic));
    }

    @Bean
    public ReactiveKafkaConsumerTemplate<String, FakeConsumerDTO> reactiveKafkaConsumerTemplate(
            ReceiverOptions<String, FakeConsumerDTO> kafkaReceiverOptions) {
        return new ReactiveKafkaConsumerTemplate<>(kafkaReceiverOptions);
    }
}
```

Q- next i have create the Configuration  For the Producer Part oki like this 

```java
package com.amir.kafka_SpringBoot.Config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate;
import org.springframework.kafka.support.serializer.JsonSerializer;
import reactor.kafka.sender.SenderOptions;

import java.util.HashMap;
import java.util.Map;

import com.amir.kafka_SpringBoot.dto.FakeProducerDTO;

@Configuration
public class ReactiveKafkaProducerConfig {
    
    @Value("${FAKE_PRODUCER_DTO_TOPIC}")
    private String topic;

    @Bean
    public ReactiveKafkaProducerTemplate<String, FakeProducerDTO> reactiveKafkaProducerTemplate(
            KafkaProperties properties) {
        Map<String, Object> props = new HashMap<>(properties.buildProducerProperties());
        
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        SenderOptions<String, FakeProducerDTO> senderOptions = SenderOptions.create(props);
        return new ReactiveKafkaProducerTemplate<>(senderOptions);
    }
}
```

Q- and finally for this part is the [application.properties](http://application.properties) 

```
spring.application.name=kafka-SpringBoot
# Kafka Broker Address
spring.kafka.bootstrap-servers=localhost:9092

# Producer Configuration
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

# Consumer Configuration
spring.kafka.consumer.group-id=reactivekafkaconsumerandproducer
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer

# JSON Deserializer Configuration
spring.kafka.properties.spring.json.trusted.packages=*
spring.kafka.consumer.properties.spring.json.use.type.headers=false
spring.kafka.consumer.properties.spring.json.value.default.type=com.example.reactivekafkaconsumerandproducer.dto.FakeConsumerDTO

# Kafka Topics
FAKE_PRODUCER_DTO_TOPIC=fake_producer_dto_topic
FAKE_CONSUMER_DTO_TOPIC=fake_consumer_dto_topic

```

Q- now we need to fix some errors like this 

The method buildConsumerProperties() from the type KafkaProperties has been deprecated since version 3.2.0 and marked for removal

and this 

The method buildProducerProperties() from the type KafkaProperties has been deprecated since version 3.2.0 and marked for removal

Q- how to fix it 

Q- this is the answer of Gemini 

The buildConsumerProperties() and buildProducerProperties() methods were deprecated because Spring Boot now favors using dedicated ConsumerFactory and ProducerFactory beans for more flexibility and control.  Here's how to fix your configuration classes:

      `package com.amir.kafka_SpringBoot.Config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.sender.SenderOptions;

import com.amir.kafka_SpringBoot.dto.FakeConsumerDTO;
import com.amir.kafka_SpringBoot.dto.FakeProducerDTO;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class ReactiveKafkaConfig { // Combined config

    @Value("${FAKE_PRODUCER_DTO_TOPIC}")
    private String producerTopic;

    @Value("${FAKE_CONSUMER_DTO_TOPIC}")
    private String consumerTopic;

    @Bean
    public ReactiveKafkaConsumerTemplate<String, FakeConsumerDTO> reactiveKafkaConsumerTemplate(
            KafkaProperties kafkaProperties) {

        Map<String, Object> consumerProps = kafkaProperties.buildConsumerProperties(); // Base properties

        // Add/override specific properties
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        consumerProps.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        consumerProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, FakeConsumerDTO.class); // Use class directly
        consumerProps.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, false);

        ReceiverOptions<String, FakeConsumerDTO> receiverOptions = ReceiverOptions.create(consumerProps)
                .subscription(Collections.singletonList(consumerTopic));

        return new ReactiveKafkaConsumerTemplate<>(receiverOptions);
    }

    @Bean
    public ReactiveKafkaProducerTemplate<String, FakeProducerDTO> reactiveKafkaProducerTemplate(
            KafkaProperties properties) {
        Map<String, Object> producerProps = properties.buildProducerProperties();

        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        SenderOptions<String, FakeProducerDTO> senderOptions = SenderOptions.create(producerProps);

        return new ReactiveKafkaProducerTemplate<>(senderOptions);
    }
}`

content_copy  Use code [with caution](https://support.google.com/legal/answer/13505487).Java

**Key Changes and Explanations:**

1. **Combined Configuration:** It's generally cleaner to keep related Kafka configurations (consumer and producer) in a single class. I've renamed the class to ReactiveKafkaConfig.
2. **Direct Class Reference:** Instead of using a string for JsonDeserializer.VALUE_DEFAULT_TYPE, use the FakeConsumerDTO.class directly. This is type-safe and avoids potential errors.
3. **Clearer Property Names:** I've made the topic names in the @Value annotations more descriptive ( producerTopic, consumerTopic).
4. **Simplified Property Handling:** You don't need to create a completely new HashMap. Use kafkaProperties.buildConsumerProperties() and kafkaProperties.buildProducerProperties() to get the base properties and then add or override specific settings as needed. This makes your configuration more maintainable.

**application.properties (No changes required for this fix but good practice):**

      `spring.application.name=kafka-SpringBoot
spring.kafka.bootstrap-servers=localhost:9092

# You can remove the explicitly set serializer/deserializer properties here 
# as they are now being set in the Java config, keeping things more centralized.
# Leaving them won't cause harm but can lead to confusion if you modify them 
# in only one place.  For better maintainability, remove them:

#spring.kafka.producer.key-serializer=...  (REMOVE)
#spring.kafka.producer.value-serializer=... (REMOVE)
#spring.kafka.consumer.key-deserializer=... (REMOVE)
#spring.kafka.consumer.value-deserializer=... (REMOVE)

spring.kafka.consumer.group-id=reactivekafkaconsumerandproducer
spring.kafka.consumer.auto-offset-reset=earliest

# It's recommended to keep the trusted packages and JSON type handling in 
# the Java config for consistency. You can REMOVE these from application.properties:
#spring.kafka.properties.spring.json.trusted.packages=* (REMOVE)
#spring.kafka.consumer.properties.spring.json.use.type.headers=false (REMOVE)
#spring.kafka.consumer.properties.spring.json.value.default.type=... (REMOVE)

FAKE_PRODUCER_DTO_TOPIC=fake_producer_dto_topic
FAKE_CONSUMER_DTO_TOPIC=fake_consumer_dto_topic`

content_copy  Use code [with caution](https://support.google.com/legal/answer/13505487).Properties

By making these changes, you're using the recommended approach for Kafka configuration in Spring Boot 3 and making your code more robust and maintainable.

Q- did this is correct ?