**Description:**

Create a microservices-based application using Spring Boot, where two separate microservices communicate asynchronously through Kafka. This project will help you understand the fundamentals of microservices architecture, Spring Boot integration with Kafka, and inter-service communication patterns.

**Components:**

- **Microservice A:** Acts as a producer that sends messages/events to a Kafka topic. For example, a "User Service" that publishes user registration events.
- **Microservice B:** Acts as a consumer that listens to the Kafka topic and processes the incoming messages. For example, a "Notification Service" that sends welcome emails when a new user registers.
- **Kafka Broker:** Facilitates the message exchange between the two microservices.
- **Spring Boot:** Framework used to build both microservices with easy integration to Kafka using Spring Kafka.

**Key Learning Points:**

- Setting up Spring Boot applications with Kafka dependencies.
- Configuring Kafka producers and consumers in Spring Boot.
- Understanding topics, partitions, and consumer groups.
- Handling message serialization/deserialization.
- Implementing error handling and retries.

Q- step 1 : create a docker compose file 

```
version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.1
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.3.1
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  schema-registry:
    image: confluentinc/cp-schema-registry:7.3.1
    depends_on:
      - kafka
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
```

Q- next run it 

docker-compose up -d 

Q- step 2 : step up a user-service appliction 

this the pom.xml file okii

```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>user-service</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>user-service</name>
    <description>Demo to learn user service</description>
    <url/>
    <licenses>
        <license/>
    </licenses>
    <developers>
        <developer/>
    </developers>
    <scm>
        <connection/>
        <developerConnection/>
        <tag/>
        <url/>
    </scm>
    <properties>
        <java.version>17</java.version>
        <confluent.version>7.3.1</confluent.version>
        <avro.version>1.11.4</avro.version>
    </properties>

    <repositories>
        <repository>
            <id>confluent</id>
            <url>https://packages.confluent.io/maven/</url>
        </repository>
    </repositories>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <!-- Confluent Schema Registry and Avro dependencies -->
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-schema-registry-client</artifactId>
            <version>${confluent.version}</version>
        </dependency>
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-avro-serializer</artifactId>
            <version>${confluent.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro</artifactId>
            <version>${avro.version}</version>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-boot-starter</artifactId>
            <version>3.0.0</version>
        </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>
    </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- and this is the application.yml for it 

```
spring:

  application:
    name: user-service
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
    properties:
      schema.registry.url: http://localhost:8081
server:
  port: 8081
```

Q- next this is : notification-service 

```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>notification-service</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>notification-service</name>
    <description>Demo project for Spring Boot</description>
    <url/>
    <licenses>
        <license/>
    </licenses>
    <developers>
        <developer/>
    </developers>
    <scm>
        <connection/>
        <developerConnection/>
        <tag/>
        <url/>
    </scm>
    <properties>
        <java.version>17</java.version>
        <confluent.version>7.3.1</confluent.version>
        <avro.version>1.11.4</avro.version>
    </properties>

    <repositories>
        <repository>
            <id>confluent</id>
            <url>https://packages.confluent.io/maven/</url>
        </repository>
    </repositories>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <!-- Confluent Schema Registry and Avro dependencies -->
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-schema-registry-client</artifactId>
            <version>${confluent.version}</version>
        </dependency>
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-avro-serializer</artifactId>
            <version>${confluent.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro</artifactId>
            <version>${avro.version}</version>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-boot-starter</artifactId>
            <version>3.0.0</version>
        </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>
    </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- and this the application.yml 

```
spring:
  application:
    name: notification-service
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: notification-service-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      properties:
        schema.registry.url: http://localhost:8081
        specific.avro.reader: true
```

Q- give this done okii 

Q- what is the next steps in details 

![image.png](https://prod-files-secure.s3.us-west-2.amazonaws.com/c26cbd8c-4ae9-4ab6-b595-99bb653f9d83/04405020-2e59-4db3-a160-6827cb03ac9d/image.png)

Q- bahii honii bsh nmchiw le path user-service/src/main/resources/avro/user-registration.avsc

Q- next try to write this 

![image.png](https://prod-files-secure.s3.us-west-2.amazonaws.com/c26cbd8c-4ae9-4ab6-b595-99bb653f9d83/86abe209-2ead-40f8-bb7d-0fb27f120b78/image.png)

### **Step 3: Define the Avro Schema**

Before sending messages between microservices, it's essential to define a schema that both services will agree upon. We'll use Avro for schema definition.

**Create a Directory for Schemas:**
In your project root (you can choose either **`user-service`** or **`notification-service`**), create a directory to store your Avro schemas.

```
user-service/
└── src/
    └── main/
        └── resources/
            └── avro/
                └── user-registration.avsc

```

**Define the Avro Schema:**
Create a file named **`user-registration.avsc`** with the following content:

```json
{
  "namespace": "com.amir.avro",
  "type": "record",
  "name": "UserRegistration",
  "fields": [
    {"name": "id", "type": "string"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": "string"},
    {"name": "registrationDate", "type": "long"}
  ]
}

```

**Explanation:**

- **Namespace:** Organizes your schema.
- **Record Name:** **`UserRegistration`** represents a user registration event.
- **Fields:** Define the data structure for the event.

### **Step 4: Generate Java Classes from Avro Schema**

To work with Avro schemas in Java, you need to generate Java classes from the **`.avsc`** files.

**Add Avro Maven Plugin to `pom.xml`:**
Update the **`pom.xml`** in both **`user-service`** and **`notification-service`** to include the Avro Maven plugin for generating Java classes.

```xml
<build>
    <plugins>
        <plugin>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-maven-plugin</artifactId>
            <version>1.11.4</version>
            <executions>
                <execution>
                    <phase>generate-sources</phase>
                    <goals>
                        <goal>schema</goal>
                    </goals>
                    <configuration>
                        <sourceDirectory>${project.basedir}/src/main/resources/avro</sourceDirectory>
                        <outputDirectory>${project.basedir}/src/main/java</outputDirectory>
                    </configuration>
                </execution>
            </executions>
        </plugin>
        <!-- Existing Spring Boot Maven Plugin -->
        <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>

```

**Generate the Java Classes:**
Run the following Maven command in both **`user-service`** and **`notification-service`** to generate the Avro Java classes:

```
mvn clean compile

```

This will generate the **`UserRegistration.java`** class in **`src/main/java/com/amir/avro/`**.

Q- i have one question the defintion of 

{
"namespace": "com.amir.avro",
"type": "record",
"name": "UserRegistration",
"fields": [
{"name": "id", "type": "string"},
{"name": "name", "type": "string"},
{"name": "email", "type": "string"},
{"name": "registrationDate", "type": "long"} // No trailing comma here
]
}

Q- can be only in one service right ? 

You are absolutely correct. The Avro schema definition (the .avsc file) needs to be shared between the services, not duplicated. It acts as a contract.

Q- then i create this 

### **Step 5: Implement the User Service (Producer)**

The User Service will expose a REST API to create a new user, which will then publish a **`UserRegistration`** event to Kafka.

**Create the `UserRegistration` Model:**
The Avro plugin should have generated a class similar to this:

```java
package com.amir.avro;

import org.apache.avro.specific.SpecificRecordBase;
import org.apache.avro.Schema;

public class UserRegistration extends SpecificRecordBase {
    private String id;
    private String name;
    private String email;
    private long registrationDate;

    // Getters and Setters
    // ...
}

```

**Create a REST Controller:**
Create a REST controller to handle user registration.

```java
package com.amir.userservice.controller;

import com.amir.avro.UserRegistration;
import com.amir.userservice.service.UserProducerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.time.Instant;
import java.util.UUID;

@RestController
@RequestMapping("/users")
public class UserController {

    @Autowired
    private UserProducerService userProducerService;

    @PostMapping("/register")
    public ResponseEntity<String> registerUser(@RequestBody UserDTO userDTO) {
        UserRegistration user = new UserRegistration();
        user.setId(UUID.randomUUID().toString());
        user.setName(userDTO.getName());
        user.setEmail(userDTO.getEmail());
        user.setRegistrationDate(Instant.now().toEpochMilli());

        userProducerService.sendUserRegistration(user);

        return ResponseEntity.ok("User registered successfully");
    }
}

// DTO Class
class UserDTO {
    private String name;
    private String email;

    // Getters and Setters
}

```

**Implement the Producer Service:**
This service handles sending messages to Kafka.

```java
package com.amir.userservice.service;

import com.amir.avro.UserRegistration;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class UserProducerService {

    private static final String TOPIC = "user-registration";

    @Autowired
    private KafkaTemplate<String, UserRegistration> kafkaTemplate;

    public void sendUserRegistration(UserRegistration user) {
        kafkaTemplate.send(TOPIC, user.getId(), user);
    }
}

```

**Configure Kafka Producer:**
Ensure that **`application.yml`** is correctly set to use Avro serialization.

```yaml
spring:

  application:
    name: user-service

  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      properties:
        schema.registry.url: http://localhost:8081

server:
  port: 8081

```

**Add Kafka Template Configuration:**
Create a configuration class to define the **`KafkaTemplate`** bean.

```java
package com.amir.userservice.config;

import com.amir.avro.UserRegistration;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

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

@Configuration
public class KafkaProducerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.properties.schema.registry.url}")
    private String schemaRegistryUrl;

    @Bean
    public ProducerFactory<String, UserRegistration> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        configProps.put("schema.registry.url", schemaRegistryUrl);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, UserRegistration> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

```

**Run the User Service:**
Start the **`user-service`** application. Ensure there are no errors and that it connects to Kafka and Schema Registry successfully.

Q- next i get this 

### **Step 6: Implement the Notification Service (Consumer)**

The Notification Service will listen to the **`user-registration`** topic and process incoming **`UserRegistration`** events by sending a welcome email (simulated).

**Create a Consumer Configuration:**
Define how the Notification Service will consume messages from Kafka.

```java
package com.amir.notificationservice.config;

import com.amir.avro.UserRegistration;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

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

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Value("${spring.kafka.properties.schema.registry.url}")
    private String schemaRegistryUrl;

    @Bean
    public ConsumerFactory<String, UserRegistration> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
        props.put("schema.registry.url", schemaRegistryUrl);
        props.put("specific.avro.reader", "true");
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, UserRegistration> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, UserRegistration> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

```

Q- but as you khnow i only create the avro User 

UserRegistration.avro in the UserService 

but in the notifcation service i get this error 

[{
"resource": "/c:/Users/PCS/Desktop/Github/Kafka-HandsOn/notification-service/src/main/java/com/amir/notification_service/config/KafkaConsumerConfig.java",
"owner": "*generated_diagnostic_collection_name*#5",
"code": "268435846",
"severity": 8,
"message": "The import com.amir.avro cannot be resolved",
"source": "Java",
"startLineNumber": 16,
"startColumn": 8,
"endLineNumber": 16,
"endColumn": 21
}]

UserRegistration cannot be resolved to a type

Q- how to fix this error