KafkaIQ

10.2.4 – Strongly-Typed Producers and Consumers

Module: Section 10 – Kafka Ecosystem & Tools
Section: 10.2 – Avro & Strongly Typed Messaging
Goal: Build production-ready, type-safe Kafka producers and consumers using Avro SpecificRecord

Introduction

This module demonstrates best practices for building strongly-typed Kafka applications with Avro, including error handling, validation, and production patterns.

Why This Module Matters: Production systems require type safety, comprehensive error handling, and maintainability—SpecificRecord delivers all three.

Production Producer Setup

Complete Producer Implementation

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import com.example.avro.User;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.Future;

public class UserProducer {
    
    private static final Logger logger = LoggerFactory.getLogger(UserProducer.class);
    private final KafkaProducer<String, User> producer;
    private final String topic;
    
    public UserProducer(String bootstrapServers, String schemaRegistryUrl, String topic) {
        this.topic = topic;
        this.producer = createProducer(bootstrapServers, schemaRegistryUrl);
    }
    
    private KafkaProducer<String, User> createProducer(String bootstrapServers, String schemaRegistryUrl) {
        Properties props = new Properties();
        
        // Kafka broker configuration
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "user-producer");
        
        // Serialization
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
        
        // Schema Registry
        props.put("schema.registry.url", schemaRegistryUrl);
        props.put("auto.register.schemas", "false");  // ✅ Manual registration in prod
        props.put("use.latest.version", "true");
        
        // Reliability configuration
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
        
        // Performance tuning
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        
        // Retries
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        
        return new KafkaProducer<>(props);
    }
    
    /**
     * Synchronous send with error handling
     */
    public void sendSync(User user) throws Exception {
        String key = String.valueOf(user.getUserId());
        ProducerRecord<String, User> record = new ProducerRecord<>(topic, key, user);
        
        try {
            RecordMetadata metadata = producer.send(record).get();
            logger.info("Sent user {} to partition {}, offset {}", 
                user.getUserId(), metadata.partition(), metadata.offset());
        } catch (Exception e) {
            logger.error("Failed to send user {}: {}", user.getUserId(), e.getMessage(), e);
            throw e;
        }
    }
    
    /**
     * Asynchronous send with callback
     */
    public Future<RecordMetadata> sendAsync(User user, Callback callback) {
        String key = String.valueOf(user.getUserId());
        ProducerRecord<String, User> record = new ProducerRecord<>(topic, key, user);
        
        return producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                logger.debug("Sent user {} to partition {}, offset {}", 
                    user.getUserId(), metadata.partition(), metadata.offset());
                if (callback != null) {
                    callback.onCompletion(metadata, null);
                }
            } else {
                logger.error("Failed to send user {}: {}", 
                    user.getUserId(), exception.getMessage(), exception);
                if (callback != null) {
                    callback.onCompletion(null, exception);
                }
            }
        });
    }
    
    /**
     * Fire-and-forget send (use with caution)
     */
    public void sendFireAndForget(User user) {
        String key = String.valueOf(user.getUserId());
        ProducerRecord<String, User> record = new ProducerRecord<>(topic, key, user);
        producer.send(record);  // No error handling!
    }
    
    public void close() {
        logger.info("Closing producer");
        producer.close();
    }
}

Usage Example

public class UserService {
    
    public static void main(String[] args) {
        UserProducer producer = new UserProducer(
            "localhost:9092",
            "http://localhost:8081",
            "users"
        );
        
        try {
            // Create type-safe user
            User user = User.newBuilder()
                .setUserId(123L)
                .setUsername("alice")
                .setEmail("alice@example.com")
                .build();
            
            // Synchronous send
            producer.sendSync(user);
            
            // Asynchronous send with callback
            producer.sendAsync(user, (metadata, exception) -> {
                if (exception == null) {
                    System.out.println("Success: " + metadata.offset());
                } else {
                    System.err.println("Failed: " + exception.getMessage());
                }
            });
            
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}

Production Consumer Setup

Complete Consumer Implementation

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import com.example.avro.User;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class UserConsumer {
    
    private static final Logger logger = LoggerFactory.getLogger(UserConsumer.class);
    private final KafkaConsumer<String, User> consumer;
    private final String topic;
    private volatile boolean running = true;
    
    public UserConsumer(String bootstrapServers, String schemaRegistryUrl, 
                        String groupId, String topic) {
        this.topic = topic;
        this.consumer = createConsumer(bootstrapServers, schemaRegistryUrl, groupId);
    }
    
    private KafkaConsumer<String, User> createConsumer(String bootstrapServers, 
                                                        String schemaRegistryUrl, 
                                                        String groupId) {
        Properties props = new Properties();
        
        // Kafka broker configuration
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "user-consumer");
        
        // Deserialization
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
        
        // Schema Registry
        props.put("schema.registry.url", schemaRegistryUrl);
        props.put("specific.avro.reader", "true");  // ✅ Use SpecificRecord
        
        // Offset management
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");  // Manual commits
        
        // Performance tuning
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024);
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        
        // Session management
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
        
        return new KafkaConsumer<>(props);
    }
    
    /**
     * Start consuming with manual commit
     */
    public void start(UserMessageHandler handler) {
        consumer.subscribe(Collections.singletonList(topic));
        logger.info("Started consumer for topic: {}", topic);
        
        try {
            while (running) {
                ConsumerRecords<String, User> records = consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, User> record : records) {
                    try {
                        // Type-safe message handling
                        User user = record.value();
                        logger.debug("Received user {} from partition {}, offset {}", 
                            user.getUserId(), record.partition(), record.offset());
                        
                        // Process message
                        handler.handle(user, record);
                        
                    } catch (Exception e) {
                        logger.error("Error processing record at offset {}: {}", 
                            record.offset(), e.getMessage(), e);
                        // Decide: skip, retry, or dead-letter queue
                    }
                }
                
                // Manual commit after processing batch
                if (!records.isEmpty()) {
                    consumer.commitSync();
                    logger.debug("Committed offsets for {} records", records.count());
                }
            }
        } catch (Exception e) {
            logger.error("Consumer error: {}", e.getMessage(), e);
        } finally {
            consumer.close();
            logger.info("Consumer closed");
        }
    }
    
    /**
     * Graceful shutdown
     */
    public void stop() {
        logger.info("Stopping consumer");
        running = false;
    }
    
    /**
     * Functional interface for message handling
     */
    @FunctionalInterface
    public interface UserMessageHandler {
        void handle(User user, ConsumerRecord<String, User> record) throws Exception;
    }
}

Usage Example

public class UserConsumerApp {
    
    public static void main(String[] args) {
        UserConsumer consumer = new UserConsumer(
            "localhost:9092",
            "http://localhost:8081",
            "user-consumer-group",
            "users"
        );
        
        // Graceful shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(consumer::stop));
        
        // Start consuming with type-safe handler
        consumer.start((user, record) -> {
            // Type-safe field access
            long userId = user.getUserId();
            String username = user.getUsername();
            String email = user.getEmail();
            
            // Business logic
            System.out.printf("Processing user: id=%d, username=%s, email=%s\n", 
                userId, username, email);
            
            // Example: Save to database
            // userRepository.save(user);
        });
    }
}

Error Handling Patterns

1. Schema Compatibility Errors

try {
    producer.send(record).get();
} catch (ExecutionException e) {
    if (e.getCause() instanceof SerializationException) {
        logger.error("Schema incompatibility detected: {}", e.getMessage());
        // Schema Registry rejected the schema
        // Action: Fix schema or rollback code
    }
}

2. Schema Registry Unavailable

try {
    producer.send(record).get();
} catch (ExecutionException e) {
    if (e.getCause() instanceof RestClientException) {
        logger.error("Schema Registry unreachable: {}", e.getMessage());
        // Action: Alert, retry, or use circuit breaker
    }
}

3. Null Field Handling

User user = record.value();
String email = user.getEmail();  // May return null

if (email != null && !email.isEmpty()) {
    sendEmailNotification(email);
} else {
    logger.warn("User {} has no email", user.getUserId());
}

4. Dead Letter Queue Pattern

public void processWithDLQ(User user, ConsumerRecord<String, User> record) {
    try {
        processUser(user);
    } catch (Exception e) {
        logger.error("Failed to process user {}: {}", user.getUserId(), e.getMessage());
        
        // Send to DLQ
        ProducerRecord<String, User> dlqRecord = new ProducerRecord<>(
            "users-dlq", 
            record.key(), 
            user
        );
        dlqProducer.send(dlqRecord);
    }
}

Validation Patterns

Schema-Level Validation

public boolean isValidUser(User user) {
    // Avro schema ensures type correctness
    // Add business logic validation
    if (user.getUserId() <= 0) {
        logger.warn("Invalid userId: {}", user.getUserId());
        return false;
    }
    
    if (user.getUsername() == null || user.getUsername().isEmpty()) {
        logger.warn("Empty username for user {}", user.getUserId());
        return false;
    }
    
    return true;
}

// Before sending
if (isValidUser(user)) {
    producer.sendSync(user);
} else {
    logger.error("Validation failed for user: {}", user);
}

Testing with SpecificRecord

Unit Test with MockProducer

import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.Test;

import static org.junit.jupiter.api.Assertions.*;

public class UserProducerTest {
    
    @Test
    public void testProduceUser() {
        // Create mock producer
        MockProducer<String, User> mockProducer = new MockProducer<>(
            true,  // autoComplete
            new StringSerializer(),
            new KafkaAvroSerializer()  // Note: Requires Schema Registry in test
        );
        
        // Create user
        User user = User.newBuilder()
            .setUserId(123L)
            .setUsername("alice")
            .setEmail("alice@example.com")
            .build();
        
        // Send
        mockProducer.send(new ProducerRecord<>("users", "123", user));
        
        // Verify
        assertEquals(1, mockProducer.history().size());
        ProducerRecord<String, User> sent = mockProducer.history().get(0);
        assertEquals("123", sent.key());
        assertEquals(123L, sent.value().getUserId());
    }
}

Integration Test with Testcontainers

import org.junit.jupiter.api.Test;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.utility.DockerImageName;

public class UserProducerIntegrationTest {
    
    @Test
    public void testProduceToRealKafka() {
        // Start Kafka & Schema Registry in Docker
        try (KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.5.0"));
             GenericContainer<?> schemaRegistry = new GenericContainer<>(DockerImageName.parse("confluentinc/cp-schema-registry:7.5.0"))
                 .withExposedPorts(8081)
                 .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "PLAINTEXT://kafka:9092")) {
            
            kafka.start();
            schemaRegistry.start();
            
            // Create producer
            UserProducer producer = new UserProducer(
                kafka.getBootstrapServers(),
                "http://" + schemaRegistry.getHost() + ":" + schemaRegistry.getFirstMappedPort(),
                "users"
            );
            
            // Send message
            User user = User.newBuilder()
                .setUserId(123L)
                .setUsername("alice")
                .setEmail("alice@example.com")
                .build();
            
            producer.sendSync(user);
            
            // Assert (would normally verify consumer received it)
        }
    }
}

Key Takeaways

  1. Production Producer: Includes idempotence, retries, compression, and error handling.

  2. Production Consumer: Manual offset commits, graceful shutdown, dead-letter queue pattern.

  3. Type Safety: SpecificRecord provides compile-time safety, IDE autocomplete, and builder pattern.

  4. Error Handling: Catch SerializationException (schema compatibility) and RestClientException (Schema Registry).

  5. Validation: Combine Avro schema validation with business logic validation.

  6. Configuration: Set auto.register.schemas=false in production, specific.avro.reader=true for consumers.

  7. Testing: Use MockProducer/MockConsumer for unit tests, Testcontainers for integration tests.

  8. Dead Letter Queue: Send failed messages to DLQ for later inspection/replay.

  9. Graceful Shutdown: Use shutdown hooks to close producer/consumer cleanly.

  10. Logging: Log schema errors, validation failures, and processing errors for debugging.

What's Next?

Next Module: 10.3 – Schema Evolution & Compatibility

What We Covered:

  • ✅ Production producer setup
  • ✅ Production consumer setup
  • ✅ Error handling patterns
  • ✅ Validation patterns
  • ✅ Testing strategies

What's Coming:

  • ❌ Schema evolution concepts
  • ❌ Compatibility types
  • ❌ Migration strategies

References:

  • SpecificRecord → 10.2.3: SpecificRecord vs GenericRecord
  • Schema evolution → 10.3: Schema Evolution & Compatibility
  • Operations → 10.4: Schema Registry Operations