Real-time Data Stream Processing: The Speed Demon

A focused guide to handling continuous data streams in Java applications

It’s 9 AM on a busy Monday. Your real-time analytics dashboard is showing:

Data Processing Rate: 50,000 records/second
Processing Latency: 2.5 seconds (target: <100ms)
Backlog: 2.3 million records
Memory Usage: 95%
Error Rate: 15%

“This can’t be right,” you think. “We’re supposed to process 100,000 records per second with sub-100ms latency. What’s going wrong?”

Full story for non-members | Grab My Microservices E-Book | Youtube | LinkedIn | Book a 1:1 Meeting

Welcome to the world of real-time data stream processing! ☕

The Problem: Understanding Stream Processing

Real-time stream processing is fundamentally different from batch processing:

Batch Processing vs Stream Processing

• Batch: Process data in chunks, high throughput, higher latency
• Stream: Process data as it arrives, lower latency, complex backpressure handling

Key Challenges:

1. Data arrives faster than processing capacity
2. Unpredictable data patterns
3. Memory management for unbounded streams
4. Error handling and recovery
5. Backpressure management

The Root Causes: What Goes Wrong

Cause #1: Blocking I/O Operations

// BAD: Blocking I/O in stream processing
public class StreamProcessor {
    public void processStream(Stream<String> dataStream) {
        dataStream.forEach(record -> {
            // This blocks the entire stream!
            String result = callExternalAPI(record); // 500ms blocking call
            saveToDatabase(result);
        });
    }

    private String callExternalAPI(String data) {
        // Synchronous HTTP call - blocks thread
        return restTemplate.postForObject("http://api.example.com/process", data, String.class);
    }
}

Cause #2: No Backpressure Handling

// BAD: No backpressure handling
public class StreamProcessor {
    private final Queue<String> processingQueue = new LinkedList<>();

    public void processStream(Stream<String> dataStream) {
        dataStream.forEach(record -> {
            processingQueue.offer(record); // Queue grows indefinitely!
            processRecord(record);
        });
    }
}

Cause #3: Memory Leaks in Streams

// BAD: Memory leak in stream processing
public class StreamProcessor {
    private final List<String> processedRecords = new ArrayList<>();

    public void processStream(Stream<String> dataStream) {
        dataStream.forEach(record -> {
            String processed = processRecord(record);
            processedRecords.add(processed); // Never cleared!
        });
    }
}

The Solution: Proper Stream Processing

Solution #1: Asynchronous Processing

// GOOD: Asynchronous stream processing
public class AsyncStreamProcessor {
    private final ExecutorService executor = Executors.newFixedThreadPool(10);
    private final RestTemplate restTemplate = new RestTemplate();

    public void processStream(Stream<String> dataStream) {
        dataStream.forEach(record -> {
            CompletableFuture.supplyAsync(() -> {
                return callExternalAPI(record);
            }, executor)
            .thenAccept(this::saveToDatabase)
            .exceptionally(throwable -> {
                logger.error("Error processing record: " + record, throwable);
                return null;
            });
        });
    }

    private String callExternalAPI(String data) {
        return restTemplate.postForObject("http://api.example.com/process", data, String.class);
    }
}

Solution #2: Backpressure Handling

// GOOD: Backpressure handling with bounded queues
public class BackpressureStreamProcessor {
    private final BlockingQueue<String> processingQueue = new LinkedBlockingQueue<>(1000);
    private final AtomicBoolean processing = new AtomicBoolean(true);
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

public void processStream(Stream<String> dataStream) {
        // Start consumer thread
        startConsumer();
        // Process stream with backpressure
        dataStream.forEach(record -> {
            try {
                // Block if queue is full (backpressure)
                processingQueue.offer(record, 100, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("Interrupted while offering record", e);
            }
        });
    }
    private void startConsumer() {
        executor.submit(() -> {
            while (processing.get()) {
                try {
                    String record = processingQueue.poll(1, TimeUnit.SECONDS);
                    if (record != null) {
                        processRecord(record);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });
    }
    // Dummy processRecord method for completeness
    private void processRecord(String record) {
        // Implement your record processing logic here
    }
}

Solution #3: Memory-Efficient Processing

// GOOD: Memory-efficient stream processing
public class MemoryEfficientStreamProcessor {
    private final AtomicLong processedCount = new AtomicLong(0);
    private final AtomicLong errorCount = new AtomicLong(0);

    public void processStream(Stream<String> dataStream) {
        dataStream
            .peek(record -> {
                // Log progress without storing data
                if (processedCount.incrementAndGet() % 10000 == 0) {
                    logger.info("Processed {} records", processedCount.get());
                }
            })
            .map(this::processRecord)
            .filter(Objects::nonNull)
            .forEach(this::saveToDatabase);
    }

    private String processRecord(String record) {
        try {
            // Process record without storing intermediate results
            return transformRecord(record);
        } catch (Exception e) {
            errorCount.incrementAndGet();
            logger.error("Error processing record", e);
            return null;
        }
    }
}

The Advanced: Reactive Stream Processing

Reactive Streams with Project Reactor

// GOOD: Reactive stream processing
@Component
public class ReactiveStreamProcessor {
    private final WebClient webClient;
    private final MeterRegistry meterRegistry;

    public ReactiveStreamProcessor(WebClient webClient, MeterRegistry meterRegistry) {
        this.webClient = webClient;
        this.meterRegistry = meterRegistry;
    }

    public Flux<String> processStream(Flux<String> dataStream) {
        return dataStream
            .buffer(100) // Process in batches of 100
            .flatMap(this::processBatch)
            .onErrorResume(error -> {
                logger.error("Error in stream processing", error);
                return Flux.empty();
            })
            .doOnNext(result -> {
                // Update metrics
                meterRegistry.counter("stream.processed.records").increment();
            });
    }

    private Flux<String> processBatch(List<String> batch) {
        return Flux.fromIterable(batch)
            .flatMap(this::callExternalAPI)
            .filter(Objects::nonNull)
            .flatMap(this::saveToDatabase);
    }

    private Mono<String> callExternalAPI(String data) {
        return webClient.post()
            .uri("http://api.example.com/process")
            .bodyValue(data)
            .retrieve()
            .bodyToMono(String.class)
            .timeout(Duration.ofSeconds(5))
            .onErrorResume(error -> {
                logger.error("API call failed for data: " + data, error);
                return Mono.empty();
            });
    }
}

Backpressure with Reactive Streams

// GOOD: Reactive backpressure handling
@Component
public class ReactiveBackpressureProcessor {

    public Flux<String> processStreamWithBackpressure(Flux<String> dataStream) {
        return dataStream
            .onBackpressureBuffer(1000) // Buffer up to 1000 items
            .flatMap(this::processRecord, 10) // Process 10 records concurrently
            .onBackpressureDrop(dropped -> {
                logger.warn("Dropped record due to backpressure: " + dropped);
                meterRegistry.counter("stream.dropped.records").increment();
            })
            .onErrorResume(error -> {
                logger.error("Error in stream processing", error);
                return Flux.empty();
            });
    }

    private Mono<String> processRecord(String record) {
        return Mono.fromCallable(() -> {
            // Process record
            return transformRecord(record);
        })
        .subscribeOn(Schedulers.boundedElastic())
        .timeout(Duration.ofSeconds(10))
        .onErrorResume(error -> {
            logger.error("Error processing record: " + record, error);
            return Mono.empty();
        });
    }
}

The Best Practices: Stream Processing Checklist

1. Performance

• Use asynchronous processing
• Implement proper backpressure handling
• Use appropriate thread pools
• Monitor processing latency

2. Memory Management

• Avoid storing intermediate results
• Use bounded queues
• Implement proper cleanup
• Monitor memory usage

3. Error Handling

• Handle individual record errors
• Implement retry mechanisms
• Log errors appropriately
• Monitor error rates

4. Monitoring

• Track processing rates
• Monitor queue sizes
• Set up alerts
• Track error rates

5. Testing

• Unit test stream processing logic
• Test backpressure scenarios
• Test error handling
• Load test with realistic data

The Code: Complete Example

@Service
public class ProductionStreamProcessor {
    private final ExecutorService executor = Executors.newFixedThreadPool(20);
    private final BlockingQueue<String> processingQueue = new LinkedBlockingQueue<>(1000);
    private final AtomicBoolean processing = new AtomicBoolean(true);
    private final MeterRegistry meterRegistry;
    private final Logger logger = LoggerFactory.getLogger(ProductionStreamProcessor.class);

    public ProductionStreamProcessor(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        startConsumer();
    }

    public void processStream(Stream<String> dataStream) {
        dataStream.forEach(record -> {
            try {
                if (!processingQueue.offer(record, 100, TimeUnit.MILLISECONDS)) {
                    logger.warn("Queue full, dropping record: " + record);
                    meterRegistry.counter("stream.dropped.records").increment();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("Interrupted while offering record", e);
            }
        });
    }

    private void startConsumer() {
        for (int i = 0; i < 10; i++) {
            executor.submit(() -> {
                while (processing.get()) {
                    try {
                        String record = processingQueue.poll(1, TimeUnit.SECONDS);
                        if (record != null) {
                            processRecord(record);
                        }
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            });
        }
    }

    private void processRecord(String record) {
        try {
            String processed = transformRecord(record);
            saveToDatabase(processed);
            meterRegistry.counter("stream.processed.records").increment();
        } catch (Exception e) {
            logger.error("Error processing record: " + record, e);
            meterRegistry.counter("stream.error.records").increment();
        }
    }

    private String transformRecord(String record) {
        // Transform record
        return record.toUpperCase();
    }

    private void saveToDatabase(String record) {
        // Save to database
    }

    @PreDestroy
    public void shutdown() {
        processing.set(false);
        executor.shutdown();
    }
}

The Lessons: Key Takeaways

1. Asynchronous is key — Don’t block streams with synchronous operations
2. Handle backpressure — Implement proper backpressure handling
3. Monitor everything — Track processing rates, errors, and latency
4. Test thoroughly — Test with realistic data volumes
5. Plan for failure — Implement proper error handling and recovery

Remember: the best stream processing is the one that never blocks. Use asynchronous processing, handle backpressure, and monitor everything.

=========

All the stories about data processing is organised in the below list

[embed]List: Big Data Processing related Questions | Curated by Arvind Kumar | Medium Big Data Processing related Questions · Comprehensive list of questions related to data processing, there solution…codefarm0.medium.com

Follow me for more such stories and keep yourself updated with the latest tech trends.