Business Problem & Why This Solution Matters

The Challenge

In modern distributed systems, we often need to:

• Send a request and wait for a specific response
• Maintain correlation between requests and responses
• Implement timeout mechanisms for reliability
• Handle multiple concurrent requests efficiently

Traditional messaging with Kafka is fire-and-forget (asynchronous), but many business scenarios require synchronous behavior:

• Financial transactions: Need immediate confirmation
• User authentication: Must wait for validation response
• Data validation: Require real-time feedback
• API gateways: Need to aggregate responses from multiple services

The Solution

This implementation creates a synchronous abstraction layer over Kafka’s asynchronous nature by:

1. Using correlation IDs to match requests with responses
2. Creating instance-specific response topics for each client
3. Managing pending requests with CompletableFuture
4. Implementing proper timeout handling

Architecture Overview

┌─────────────────┐    request-topic    ┌─────────────────┐  
│   Kafka Client  │ ──────────────────► │  Kafka Server   │  
│                 │                     │                 │  
│ ┌─────────────┐ │                     │ ┌─────────────┐ │  
│ │Pending      │ │                     │ │Request      │ │  
│ │Requests     │ │                     │ │Processor    │ │  
│ │Map          │ │                     │ │             │ │  
│ └─────────────┘ │                     │ └─────────────┘ │  
│                 │ ◄────────────────── │                 │  
└─────────────────┘  response-topic-    └─────────────────┘  
                     {instanceId}

Key Components

KafkaSyncClient

Purpose: Sends synchronous requests and waits for responses

Key Features:

• Correlation ID Management: Each request gets a unique UUID
• Instance-Specific Response Topics: Creates response-topic-{instanceId} for each client instance
• Timeout Handling: Configurable request timeout (default: 30 seconds)
• Concurrent Request Support: Uses ConcurrentHashMap to track multiple pending requests.

Implementation Highlights

fun sendSyncRequest(request: String): String {  
    val correlationId: UUID = UUID.randomUUID()  
    val future = CompletableFuture<String>()  
    pendingRequests\[correlationId\] = future  
  
    // Send request with correlation ID and instance ID in headers  
    val headers: List<RecordHeader> = listOf(  
        RecordHeader("correlationId", correlationId.toString().toByteArray()),  
        RecordHeader("instanceId", kafkaProperties.instanceId.toByteArray())  
    )  
    val record: ProducerRecord<String, String> = ProducerRecord(  
        kafkaProperties.requestTopic,  
        null,  
        correlationId.toString(),  
        request,  
        headers  
    )  
    producer.send(record) { metadata: RecordMetadata?, exception: Exception? ->  
        log.info("Sent request to topic ${kafkaProperties.requestTopic} with correlationId: $correlationId")  
        if (exception != null) {  
            future.completeExceptionally(exception)  
            pendingRequests.remove(key = correlationId)  
        }  
    }  
    return try {  
        // Block until response is received or timeout occurs  
        future.get(kafkaProperties.requestTimeout, TimeUnit.MILLISECONDS)  
    } catch (e: Exception) {  
        pendingRequests.remove(key = correlationId)  
        log.error("Request timed out or failed. Exception: ${e.localizedMessage}")  
        throw RuntimeException("Request timed out or failed", e)  
    } finally {  
        log.info("Request with correlationId: $correlationId completed")  
        future.complete(null)  
    }  
}

Start the response consumer in a separate thread to listen for responses based on correlation IDs:

private fun consumeResponses() {  
    while (true) {  
        try {  
            val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofMillis(100))  
            for (record: ConsumerRecord<String, String> in records) {  
                log.info("Received response for correlationId: ${record.key()}")  
                val correlationId: UUID = record.headers()  
                    .first { it.key() == "correlationId" }  
                    ?.value()  
                    ?.let { String(bytes = it) }  
                    ?.let { UUID.fromString(it) }  
                    ?: continue  
                val future: CompletableFuture<String>? = pendingRequests.remove(key = correlationId)  
                future?.complete(record.value())  
            }  
        } catch (e: Exception) {  
            log.error("Error consuming responses. Exception: ${e.localizedMessage}")  
        }  
    }  
}

KafkaSyncServer

Purpose: Processes requests and sends responses back to the appropriate client.

Key Features:

• Request Processing: Listens to request-topic
• Dynamic Response Routing: Sends responses to client-specific topics
• Header Extraction: Reads correlation ID and instance ID from request headers
• Automatic Response Generation: Processes requests and generates responses.

Core Request Processing Logic

private fun consumeRequests() {  
    while (running) {  
        try {  
            // Poll for new requests every 100ms  
            val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofMillis(100))  
              
            for (record: ConsumerRecord<String, String> in records) {  
                log.info("Received request: ${record.value()} from topic: ${record.topic()}")  
                  
                // Extract correlation ID from message headers  
                val correlationId: String = record.headers()  
                    .first { it.key() == "correlationId" }  
                    ?.value()  
                    ?.let { String(it) }  
                    ?: continue // Skip messages without correlation ID  
                  
                // Extract instance ID to route response correctly  
                val instanceId: String = record.headers()  
                    .first { it.key() == "instanceId" }  
                    ?.value()  
                    ?.let { String(it) }  
                    ?: continue // Skip messages without instance ID  
                  
                // Process the actual business logic  
                val processedResponse = processBusinessLogic(record.value(), correlationId)  
                  
                // Send response to client-specific topic  
                sendResponse(instanceId, correlationId, processedResponse)  
            }  
        } catch (e: Exception) {  
            log.error("Error processing requests: ${e.localizedMessage}")  
        }  
    }  
}

Processing Flow Explained

• Polling Strategy: Uses short 100ms polls to balance responsiveness vs CPU usage
• Header Extraction: Safely extracts correlation ID and instance ID with null checks
• Error Resilience: Continues processing even if individual messages fail
• Instance Routing: Uses instance ID to determine correct response topic

Dynamic Response Routing

private fun sendResponse(instanceId: String, correlationId: String, response: String) {  
    // Construct instance-specific response topic  
    val responseTopicName = "${kafkaProperties.responseTopic}\-$instanceId"  
      
    // Create response record with correlation headers  
    val responseRecord: ProducerRecord<String, String> = ProducerRecord(  
        responseTopicName,                    // Topic: response-topic-{instanceId}  
        null,                                 // Partition: let Kafka decide  
        correlationId,                        // Key: correlation ID for ordering  
        response,                            // Value: processed response  
        listOf(                              // Headers: for client correlation  
            RecordHeader("correlationId", correlationId.toByteArray())  
        )  
    )  
      
    // Send response asynchronously with callback  
    producer.send(responseRecord) { metadata: RecordMetadata?, exception: Exception? ->  
        if (exception != null) {  
            log.error("Failed to send response for correlation ID $correlationId: ${exception.localizedMessage}")  
        } else {  
            log.info("Response sent to $responseTopicName for correlation ID: $correlationId")  
        }  
    }  
}

Response Routing Features

• Dynamic Topic Selection: Each client instance gets responses on its own topic
• Correlation Preservation: Correlation ID maintained in both key and headers
• Async Send with Callback: Non-blocking response sending with error handling
• Partition Strategy: Lets Kafka distribute responses across partitions

Configuration

Application Properties (application.yml)

kafka:  
  bootstrap-servers: localhost:9092  
  request-topic: request-topic  
  response-topic: response-topic  
  request-timeout: 30000

Docker Compose Setup

The project includes a complete Kafka setup using KRaft mode (no Zookeeper required):

kafka_kraft:  
  image: docker.io/bitnami/kafka:3.4  
  ports:  
    - "9092:9092"  
  environment:  
    - KAFKA_ENABLE_KRAFT=yes  
    - KAFKA_AUTO_CREATE_TOPICS_ENABLE=true

Advanced Features

Instance Isolation

Each application instance gets its own response topic:

• response-topic-{instanceId} that prevents response mix-ups in multi-instance deployments and enables horizontal scaling.

Correlation ID Tracking

• Every request gets a unique UUID
• Responses are matched using correlation IDs
• Failed matches are logged and ignored

When to Use This Pattern

✅ Good Use Cases

• API Gateways: Need to aggregate responses from multiple services
• Microservice Communication: When you need guaranteed responses
• Financial Systems: Transaction processing requiring immediate confirmation
• Real-time Validation: User input validation with immediate feedback

❌ When Not to Use

High Throughput Streaming: Pure async is better for high volume

Fire-and-Forget Operations: No need for response correlation

Long-Running Processes: Better suited for callback patterns

Simple Request/Response: HTTP REST might be simpler

Comparison with Alternatives

Future Enhancements

1. Dead Letter Queue Implementation: Handle failed messages
2. Circuit Breaker Pattern: Prevent cascading failures
3. Response Caching: Cache responses for identical requests
4. Metrics Integration: Add Micrometer metrics for monitoring
5. Security: Add SSL/SASL authentication
6. Schema Registry Integration: For better message evolution

Conclusion

This Kafka synchronous pattern bridges the gap between Kafka’s asynchronous nature and synchronous business requirements. It’s particularly valuable in microservice architectures where you need the benefits of Kafka (scalability, durability, ordering) while maintaining request-response semantics.

The implementation demonstrates advanced Kafka concepts including:

• Dynamic topic creation
• Header-based correlation
• Instance isolation
• Proper resource management
• Timeout handling

This pattern is production-ready and can be adapted for various enterprise scenarios requiring reliable, scalable, synchronous communication over Kafka infrastructure.

Full code example can be found in github