Producer API

Overview

Kafka's Producer API allows applications to send data into Kafka topics using Java. This lesson covers implementing a Kafka producer using Java and Gradle.

Why Java and Gradle?

Java: Well-integrated with Kafka, enterprise-level support, robust ecosystem

Gradle: Simplifies dependency management, automates builds, easy integration

Project Setup

Initialize Project

bash

1mkdir kafka-producer-demo
2cd kafka-producer-demo
3gradle init --type java-application

Project Structure

build.gradle: Main configuration file
gradle/: Gradle configuration files
gradlew / gradlew.bat: Gradle wrapper scripts
settings.gradle: Project settings
src/main/: Source code
src/test/: Test cases

Add Dependencies

Edit build.gradle:

groovy

1plugins {
2    id 'java'
3}
4
5group = 'com.example'
6version = '1.0-SNAPSHOT'
7
8repositories {
9    mavenCentral()
10}
11
12dependencies {
13    implementation 'org.apache.kafka:kafka-clients:3.5.0'
14}

Build the project:

bash

1./gradlew build

Creating a Producer

Basic Producer Class

Create src/main/java/com/example/ProducerDemo.java:

java

1package com.example;
2
3import org.apache.kafka.clients.producer.*;
4import java.util.Properties;
5
6public class ProducerDemo {
7    public static void main(String[] args) {
8        // Producer configuration
9        Properties props = new Properties();
10        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
11        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
12                 "org.apache.kafka.common.serialization.StringSerializer");
13        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
14                 "org.apache.kafka.common.serialization.StringSerializer");
15        
16        // Create producer
17        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
18        
19        // Create record
20        ProducerRecord<String, String> record = 
21            new ProducerRecord<>("payment", "key1", "value1");
22        
23        // Send message
24        producer.send(record, (metadata, exception) -> {
25            if (exception == null) {
26                System.out.println("Message sent successfully");
27            } else {
28                exception.printStackTrace();
29            }
30        });
31        
32        producer.close();
33    }
34}

Key Components

Properties Configuration:

BOOTSTRAP_SERVERS_CONFIG: Broker addresses
KEY_SERIALIZER_CLASS_CONFIG: How to serialize keys
VALUE_SERIALIZER_CLASS_CONFIG: How to serialize values

Producer Record: Message to send (topic, key, value)

Send Method: Asynchronous send with callback

Producer Configuration

Essential Settings

properties

1bootstrap.servers=localhost:9092
2key.serializer=org.apache.kafka.common.serialization.StringSerializer
3value.serializer=org.apache.kafka.common.serialization.StringSerializer
4acks=all
5retries=3

Performance Settings