1.4 – Kafka APIs and Client Landscape

Module: M1 – Kafka Fundamentals
Goal: Orient learners to Kafka APIs without deep dives

Introduction

Kafka provides multiple APIs for different use cases. This module introduces each API at a high level, explaining what it does and when to use it.

Important: We'll cover what each API does, not how to configure or tune it. Deep dives come in later modules.

The Five Kafka APIs

Kafka provides five main APIs:

Producer API

What Is the Producer API?

Producer API = Write events to Kafka topics.

Analogy: Like inserting rows into a database or writing lines to a log file.

What Does the Producer API Do?

1. Serialize events (convert to bytes)
2. Determine partition (based on key or round-robin)
3. Send events to the leader broker
4. Receive acknowledgment (event saved)

Producer API Example (Java)

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class SimpleProducer {
  public static void main(String[] args) {
    // 1. Configure producer
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

    // 2. Create producer
    KafkaProducer<String, String> producer = new KafkaProducer<>(props);

    // 3. Send event
    ProducerRecord<String, String> record = 
      new ProducerRecord<>("orders", "order-123", "{\"amount\": 99.00}");
    
    producer.send(record);

    // 4. Close producer
    producer.close();
  }
}

Key Points:

• bootstrap.servers: Kafka broker addresses
• key.serializer: How to convert key to bytes
• value.serializer: How to convert value to bytes
• ProducerRecord: Event to send (topic, key, value)

When to Use the Producer API

Use Cases:

• Web API logging user activity
• Microservice publishing events (order created, payment processed)
• IoT device sending sensor readings
• Database CDC (change data capture)

Example: Order Service writes order events to Kafka.

// Order Service
Order order = createOrder();
producer.send(new ProducerRecord<>("orders", order.getId(), orderToJson(order)));

What We're NOT Covering (Yet)

Deferred to M3: Producers:

• Serialization formats (Avro, JSON, Protobuf)
• Partitioning strategies (custom partitioners)
• Producer configs (acks, retries, compression)
• Error handling and retries
• Idempotence and transactions

Consumer API

What Is the Consumer API?

Consumer API = Read events from Kafka topics.

Analogy: Like reading rows from a database or tailing a log file.

What Does the Consumer API Do?

1. Subscribe to topics
2. Poll for new events (pull-based)
3. Deserialize events (convert from bytes)
4. Process events (business logic)
5. Commit offsets (mark as read)

Consumer API Example (Java)

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

public class SimpleConsumer {
  public static void main(String[] args) {
    // 1. Configure consumer
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("group.id", "payment-processors");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    // 2. Create consumer
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

    // 3. Subscribe to topic
    consumer.subscribe(Arrays.asList("orders"));

    // 4. Poll for events
    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
      for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Received: key=%s, value=%s, offset=%d%n", 
          record.key(), record.value(), record.offset());
      }
    }
  }
}

Key Points:

• group.id: Consumer group name (for parallel processing)
• subscribe: Which topics to read
• poll: Fetch new events (non-blocking)
• ConsumerRecord: Event read from Kafka (key, value, offset)

When to Use the Consumer API

Use Cases:

• Payment Service reading order events
• Analytics Service reading user activity
• Notification Service reading order events
• Fraud Detection reading transaction events

Example: Payment Service reads order events and processes payments.

// Payment Service
while (true) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
  for (ConsumerRecord<String, String> record : records) {
    Order order = jsonToOrder(record.value());
    processPayment(order);
  }
}

What We're NOT Covering (Yet)

Deferred to M4: Consumers:

• Consumer groups and rebalancing
• Offset management (auto-commit vs manual commit)
• Consumer configs (fetch size, session timeout)
• Error handling and retries
• Exactly-once consumption

Admin API

What Is the Admin API?

Admin API = Manage Kafka resources (topics, configs, ACLs) programmatically.

Analogy: Like using SQL DDL (CREATE TABLE, ALTER TABLE) instead of the database CLI.

What Does the Admin API Do?

Administrative Operations:

• Create, delete, describe topics
• Alter topic configurations
• List consumer groups
• Describe cluster metadata
• Manage ACLs (access control lists)
• Delete consumer group offsets

Admin API Example (Java)

import org.apache.kafka.clients.admin.*;
import java.util.*;
import java.util.concurrent.ExecutionException;

public class AdminExample {
  public static void main(String[] args) throws ExecutionException, InterruptedException {
    // 1. Configure admin client
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

    // 2. Create admin client
    AdminClient admin = AdminClient.create(props);

    // 3. Create topic
    NewTopic newTopic = new NewTopic("orders", 10, (short) 3);
    admin.createTopics(Collections.singletonList(newTopic)).all().get();

    // 4. List topics
    Set<String> topics = admin.listTopics().names().get();
    System.out.println("Topics: " + topics);

    // 5. Close admin client
    admin.close();
  }
}

Key Points:

• AdminClient: Programmatic access to Kafka admin operations
• NewTopic: Create topic with partitions and replication factor
• listTopics: Get all topic names

When to Use the Admin API

Use Cases:

• Automated topic creation (CI/CD pipelines)
• Monitoring tools (track cluster health)
• Multi-tenancy (create topics per customer)
• Configuration management (automate config changes)

Example: Automated topic provisioning.

// CI/CD pipeline
if (!admin.listTopics().names().get().contains("orders")) {
  admin.createTopics(Collections.singletonList(
    new NewTopic("orders", 10, (short) 3)
  ));
}

What We're NOT Covering (Yet)

Deferred to M6: Operations:

• Cluster monitoring
• Performance metrics
• Dynamic config changes
• Partition reassignment
• Quota management

Streams API

What Is the Streams API?

Streams API = Java library for processing event streams (real-time transformations, aggregations, joins).

Analogy: Like SQL for streaming data (filter, map, reduce, join).

Key Point: Streams API is a client library, not a separate cluster (runs in your application).

What Does the Streams API Do?

Stream Processing Operations:

• Filter: Remove unwanted events
• Map: Transform events
• Aggregate: Count, sum, average
• Join: Combine events from multiple topics
• Windowing: Group events by time window

Streams API Example (Java)

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

public class WordCountExample {
  public static void main(String[] args) {
    // 1. Configure Streams
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "word-count-app");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

    // 2. Define stream processing topology
    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, String> textLines = builder.stream("text-input");

    KTable<String, Long> wordCounts = textLines
      .flatMapValues(line -> Arrays.asList(line.toLowerCase().split("\\s+")))
      .groupBy((key, word) -> word)
      .count();

    wordCounts.toStream().to("word-count-output");

    // 3. Start Streams application
    KafkaStreams streams = new KafkaStreams(builder.build(), props);
    streams.start();
  }
}

Key Points:

• StreamsBuilder: Define stream processing topology
• KStream: Unbounded stream of events
• KTable: Changelog stream (latest value per key)
• flatMapValues, groupBy, count: Stream operations

When to Use the Streams API

Use Cases:

• Real-time analytics (count events, calculate averages)
• Data enrichment (join orders with customer data)
• Alerting (detect anomalies, send alerts)
• Aggregations (calculate rolling sums, moving averages)

Example: Real-time fraud detection.

KStream<String, Transaction> transactions = builder.stream("transactions");

transactions
  .filter((key, tx) -> tx.getAmount() > 10000)  // Large transactions
  .filter((key, tx) -> tx.getLocation().equals("unusual"))  // Unusual location
  .to("fraud-alerts");  // Send to fraud alerts topic

What We're NOT Covering (Yet)

Deferred to M5: Kafka Streams:

• Stream processing concepts (stateless vs stateful)
• Windowing and time semantics
• Joins (stream-stream, stream-table, table-table)
• State stores
• Exactly-once semantics

Connect API (Kafka Connect)

What Is Kafka Connect?

Kafka Connect = Framework for data integration between Kafka and external systems (databases, files, cloud storage).

Analogy: Like ETL tools (extract, transform, load) but for streaming data.

Key Point: Connect runs as a separate cluster (not in your application).

Kafka Connect Components

Kafka Connect Architecture

External System (MySQL) → Source Connector → Kafka Topic (users)
                                                   ↓
                                              Kafka Topic (users) → Sink Connector → External System (S3)

Example: Sync MySQL database to Kafka, then to S3.

Source Connector Example

Use Case: Import MySQL database changes into Kafka (CDC).

{
  "name": "mysql-source-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql.example.com",
    "database.port": "3306",
    "database.user": "kafka",
    "database.password": "password",
    "database.server.id": "184054",
    "database.server.name": "dbserver1",
    "table.whitelist": "inventory.customers",
    "database.history.kafka.bootstrap.servers": "localhost:9092",
    "database.history.kafka.topic": "schema-changes"
  }
}

Result: Every insert/update/delete in customers table is written to Kafka topic dbserver1.inventory.customers.

Sink Connector Example

Use Case: Export Kafka events to S3 for long-term storage.

{
  "name": "s3-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "1",
    "topics": "orders",
    "s3.region": "us-west-2",
    "s3.bucket.name": "my-kafka-data",
    "flush.size": "1000",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.json.JsonFormat"
  }
}

Result: Events from topic orders are written to S3 bucket my-kafka-data in JSON format.

When to Use Kafka Connect

Use Cases:

• Database CDC (MySQL, PostgreSQL, MongoDB → Kafka)
• Data warehousing (Kafka → Snowflake, Redshift, BigQuery)
• File systems (Kafka → S3, HDFS)
• Search (Kafka → Elasticsearch)
• Caching (Kafka → Redis)

Example: Real-time data pipeline.

MySQL → Debezium (Source) → Kafka → Snowflake Sink → Snowflake

What We're NOT Covering (Yet)

Deferred to M8: Kafka Connect:

• Connector configuration
• Single Message Transforms (SMTs)
• Error handling and dead letter queues
• Custom connector development
• Distributed vs standalone mode

Language Ecosystem

Kafka supports multiple programming languages via client libraries.

Official Clients (Maintained by Apache Kafka)

Community Clients (Third-Party)

librdkafka-Based vs Pure Clients

librdkafka: High-performance C library (used by many clients).

Advantages:

• ✅ Best performance (close to Java client)
• ✅ Battle-tested (used by Confluent-maintained clients)

Disadvantages:

• ❌ Requires native dependencies (harder to install)

Pure Clients (e.g., kafka-python, kafkajs):

• ✅ Easy to install (no native dependencies)
• ❌ Slower performance
• ❌ May lag behind Kafka features

Recommendation: Use librdkafka-based clients for production (e.g., confluent-kafka-python, confluent-kafka-go).

When to Use Each API

Decision Tree

What do you need to do?

├─ Send data to Kafka?
│  └─ Use Producer API
│
├─ Read data from Kafka?
│  └─ Use Consumer API
│
├─ Manage Kafka (create topics, etc.)?
│  └─ Use Admin API
│
├─ Process streams (filter, aggregate)?
│  └─ Use Streams API
│
└─ Integrate with external systems (database, S3)?
   └─ Use Kafka Connect

Client Library Best Practices

1. Use Official or Confluent-Maintained Clients

Good:

• Java: kafka-clients
• Python: confluent-kafka-python
• Go: confluent-kafka-go

Avoid: Unmaintained third-party clients.

2. Check Kafka Version Compatibility

Example: Kafka 3.0 client can connect to Kafka 2.8 broker (backward compatible).

Rule: Use a client version ≤ broker version (or same major version).

3. Use Serialization Formats

Good:

• Avro: Schema evolution, compact
• Protobuf: Fast, compact
• JSON: Human-readable, but larger

Avoid: Plain strings for structured data (no schema validation).

4. Monitor Client Metrics

Key Metrics:

• Producer: Record send rate, error rate
• Consumer: Lag, fetch rate
• Admin: API call latency

Tools: JMX (Java), Prometheus exporters.

Key Takeaways

1. Five APIs: Producer, Consumer, Admin, Streams, Connect.

2. Producer API: Write events to Kafka.

3. Consumer API: Read events from Kafka.

4. Admin API: Manage Kafka programmatically.

5. Streams API: Real-time stream processing (filter, map, aggregate).

6. Connect API: Data integration (database CDC, S3 export).

7. Language Support: Java (best), Python, Go, .NET, Node.js (community clients).

8. librdkafka: High-performance C library used by many clients.

What's Next?

Next Module: 1.5 – Installation and Running Kafka

What We Covered:

• ✅ Producer API (write events)
• ✅ Consumer API (read events)
• ✅ Admin API (manage Kafka)
• ✅ Streams API (process streams)
• ✅ Connect API (data integration)
• ✅ Language ecosystem
• ✅ When to use each API

What We Didn't Cover (Coming in Later Modules):

• ❌ Producer configs and tuning → M3: Producers
• ❌ Consumer groups and rebalancing → M4: Consumers
• ❌ Stream processing patterns → M5: Kafka Streams
• ❌ Connector development → M8: Kafka Connect

Kafka APIs and Client Landscape: Complete ✅