Dataflow Pipeline in Java: Complete Implementation Guide

Learn how to build robust, scalable data processing pipelines in Java with modern streaming and batch processing capabilities.

Pipeline Architecture
Core Pipeline Components
Stream Processing
Batch Processing
Windowed Operations
State Management
Fault Tolerance
Monitoring & Metrics
Spring Boot Integration

Pipeline Architecture

Dataflow Pipeline Patterns

┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Source │───▶│ Transform │───▶│ Window │───▶│ Sink │ │ │ │ │ │ & Agg │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Filter │ │ State │ │ Kafka, │ │ & Map │ │ Management │ │ DB, S3 │ └─────────────┘ └─────────────┘ └─────────────┘

Core Pipeline Components

1. Pipeline Foundation Classes

// Core pipeline interfaces public interface Pipeline<T> { String getName(); Pipeline<T> from(Source<T> source); <R> Pipeline<R> transform(Transform<T, R> transform); Pipeline<T> filter(Filter<T> filter); Pipeline<T> window(WindowStrategy<T> windowStrategy); void to(Sink<T> sink); void start(); void stop(); PipelineStats getStats(); } public interface Source<T> { String getName(); void start(SourceContext<T> context); void stop(); SourceStats getStats(); } public interface SourceContext<T> { void emit(T element); void markProcessed(String elementId); void markFailed(String elementId, Throwable error); } public interface Transform<T, R> { String getName(); R apply(T input); void onError(T input, Throwable error); } public interface Sink<T> { String getName(); void write(T element); void writeBatch(List<T> elements); void flush(); SinkStats getStats(); } // Base pipeline implementation @Component @Slf4j public class DataflowPipeline<T> implements Pipeline<T> { private final String name; private final ExecutorService executor; private final PipelineMetrics metrics; private Source<T> source; private final List<Transform<?, ?>> transforms; private Sink<T> sink; private WindowStrategy<T> windowStrategy; private volatile boolean running = false; public DataflowPipeline(String name, PipelineMetrics metrics) { this.name = name; this.metrics = metrics; this.transforms = new CopyOnWriteArrayList<>(); this.executor = Executors.newFixedThreadPool(10, new ThreadFactoryBuilder() .setNameFormat("pipeline-" + name + "-%d") .build()); } @Override public Pipeline<T> from(Source<T> source) { this.source = source; return this; } @Override @SuppressWarnings("unchecked") public <R> Pipeline<R> transform(Transform<T, R> transform) { this.transforms.add(transform); // Return a new pipeline with the transformed type DataflowPipeline<R> newPipeline = new DataflowPipeline<>(name + "-transformed", metrics); newPipeline.source = new TransformedSource<>(this, transform); return newPipeline; } @Override public Pipeline<T> filter(Filter<T> filter) { this.transforms.add(new FilterTransform<>(filter)); return this; } @Override public Pipeline<T> window(WindowStrategy<T> windowStrategy) { this.windowStrategy = windowStrategy; return this; } @Override public void to(Sink<T> sink) { this.sink = sink; } @Override public void start() { if (running) { log.warn("Pipeline {} is already running", name); return; } running = true; executor.submit(() -> { log.info("Starting pipeline: {}", name); metrics.recordPipelineStart(name); source.start(new SourceContext<T>() { @Override public void emit(T element) { processElement(element); } @Override public void markProcessed(String elementId) { metrics.recordElementProcessed(name, elementId); } @Override public void markFailed(String elementId, Throwable error) { metrics.recordElementFailed(name, elementId, error); } }); log.info("Pipeline {} started successfully", name); }); } @Override public void stop() { running = false; source.stop(); executor.shutdown(); try { if (!executor.awaitTermination(30, TimeUnit.SECONDS)) { executor.shutdownNow(); } } catch (InterruptedException e) { executor.shutdownNow(); Thread.currentThread().interrupt(); } metrics.recordPipelineStop(name); log.info("Pipeline {} stopped", name); } @Override public PipelineStats getStats() { return new PipelineStats( name, metrics.getProcessedCount(name), metrics.getFailedCount(name), metrics.getProcessingRate(name), source.getStats(), sink != null ? sink.getStats() : null ); } @SuppressWarnings("unchecked") private void processElement(T element) { if (!running) return; try { Object current = element; // Apply all transforms for (Transform transform : transforms) { current = transform.apply(current); if (current == null) { // Element was filtered out return; } } // Apply windowing if configured if (windowStrategy != null) { windowStrategy.process((T) current); } else if (sink != null) { // Write to sink sink.write((T) current); } metrics.recordElementProcessed(name, element.toString()); } catch (Exception e) { log.error("Error processing element in pipeline {}: {}", name, element, e); metrics.recordElementFailed(name, element.toString(), e); } } // Statistics classes public record PipelineStats( String pipelineName, long processedCount, long failedCount, double processingRate, SourceStats sourceStats, SinkStats sinkStats ) {} public record SourceStats( String sourceName, long elementsEmitted, long errors, double emissionRate ) {} public record SinkStats( String sinkName, long elementsWritten, long errors, double writeRate ) {} }

2. Source Implementations

// Kafka Source @Component @Slf4j public class KafkaSource<T> implements Source<T> { private final KafkaTemplate<String, String> kafkaTemplate; private final ObjectMapper objectMapper; private final Class<T> type; private final String topic; private final String groupId; private volatile boolean running = false; private KafkaConsumer<String, String> consumer; private Thread consumerThread; private final AtomicLong elementsEmitted = new AtomicLong(0); private final AtomicLong errors = new AtomicLong(0); public KafkaSource(KafkaTemplate<String, String> kafkaTemplate, ObjectMapper objectMapper, Class<T> type, String topic, String groupId) { this.kafkaTemplate = kafkaTemplate; this.objectMapper = objectMapper; this.type = type; this.topic = topic; this.groupId = groupId; } @Override public String getName() { return "kafka-source-" + topic; } @Override public void start(SourceContext<T> context) { if (running) { log.warn("Kafka source for topic {} is already running", topic); return; } running = true; consumerThread = new Thread(() -> { log.info("Starting Kafka source for topic: {}", topic); Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); consumer = new KafkaConsumer<>(props); consumer.subscribe(Collections.singletonList(topic)); while (running) { try { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { try { T element = objectMapper.readValue(record.value(), type); context.emit(element); elementsEmitted.incrementAndGet(); // Manual commit after successful processing consumer.commitSync(); } catch (Exception e) { log.error("Failed to process Kafka record from topic {}: {}", topic, record.value(), e); errors.incrementAndGet(); context.markFailed(record.key(), e); } } } catch (Exception e) { log.error("Error in Kafka consumer for topic {}", topic, e); errors.incrementAndGet(); } } consumer.close(); log.info("Kafka source for topic {} stopped", topic); }, "kafka-source-" + topic); consumerThread.start(); } @Override public void stop() { running = false; if (consumerThread != null) { try { consumerThread.join(5000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } @Override public SourceStats getStats() { return new SourceStats( getName(), elementsEmitted.get(), errors.get(), elementsEmitted.get() / (System.currentTimeMillis() / 1000.0) // Simple rate calculation ); } } // File Source @Component @Slf4j public class FileSource<T> implements Source<T> { private final String filePath; private final ObjectMapper objectMapper; private final Class<T> type; private final boolean watchForChanges; private volatile boolean running = false; private Thread fileReaderThread; private final AtomicLong elementsEmitted = new AtomicLong(0); private final AtomicLong errors = new AtomicLong(0); public FileSource(String filePath, ObjectMapper objectMapper, Class<T> type, boolean watchForChanges) { this.filePath = filePath; this.objectMapper = objectMapper; this.type = type; this.watchForChanges = watchForChanges; } @Override public String getName() { return "file-source-" + filePath; } @Override public void start(SourceContext<T> context) { if (running) { log.warn("File source for {} is already running", filePath); return; } running = true; fileReaderThread = new Thread(() -> { log.info("Starting file source for: {}", filePath); try { Path path = Paths.get(filePath); if (watchForChanges) { watchFileChanges(path, context); } else { processExistingFile(path, context); } } catch (Exception e) { log.error("Error in file source for {}", filePath, e); } log.info("File source for {} stopped", filePath); }, "file-source-" + filePath); fileReaderThread.start(); } private void processExistingFile(Path path, SourceContext<T> context) throws IOException { if (!Files.exists(path)) { log.warn("File does not exist: {}", path); return; } try (Stream<String> lines = Files.lines(path)) { lines.forEach(line -> { try { T element = objectMapper.readValue(line, type); context.emit(element); elementsEmitted.incrementAndGet(); } catch (Exception e) { log.error("Failed to parse line from file {}: {}", filePath, line, e); errors.incrementAndGet(); context.markFailed(line, e); } }); } } private void watchFileChanges(Path path, SourceContext<T> context) throws IOException { WatchService watchService = FileSystems.getDefault().newWatchService(); Path directory = path.getParent(); directory.register(watchService, StandardWatchEventKinds.ENTRY_MODIFY); long lastModified = 0; while (running) { try { WatchKey key = watchService.poll(1, TimeUnit.SECONDS); if (key != null) { for (WatchEvent<?> event : key.pollEvents()) { if (event.kind() == StandardWatchEventKinds.ENTRY_MODIFY) { Path changedFile = (Path) event.context(); if (changedFile.equals(path.getFileName())) { long currentModified = Files.getLastModifiedTime(path).toMillis(); if (currentModified > lastModified) { processNewLines(path, lastModified, context); lastModified = currentModified; } } } } key.reset(); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { log.error("Error watching file {}", path, e); } } watchService.close(); } private void processNewLines(Path path, long since, SourceContext<T> context) throws IOException { // Implementation to read only new lines since last modification // This is a simplified version try (Stream<String> lines = Files.lines(path)) { lines.forEach(line -> { try { T element = objectMapper.readValue(line, type); context.emit(element); elementsEmitted.incrementAndGet(); } catch (Exception e) { log.error("Failed to parse line from file {}: {}", filePath, line, e); errors.incrementAndGet(); } }); } } @Override public void stop() { running = false; if (fileReaderThread != null) { fileReaderThread.interrupt(); } } @Override public SourceStats getStats() { return new SourceStats( getName(), elementsEmitted.get(), errors.get(), elementsEmitted.get() / (System.currentTimeMillis() / 1000.0) ); } } // Database Source (JDBC) @Component @Slf4j public class JdbcSource<T> implements Source<T> { private final DataSource dataSource; private final String query; private final RowMapper<T> rowMapper; private final long pollIntervalMs; private volatile boolean running = false; private ScheduledExecutorService scheduler; private final AtomicLong elementsEmitted = new AtomicLong(0); private final AtomicLong errors = new AtomicLong(0); public JdbcSource(DataSource dataSource, String query, RowMapper<T> rowMapper, long pollIntervalMs) { this.dataSource = dataSource; this.query = query; this.rowMapper = rowMapper; this.pollIntervalMs = pollIntervalMs; } @Override public String getName() { return "jdbc-source-" + query.hashCode(); } @Override public void start(SourceContext<T> context) { if (running) { log.warn("JDBC source is already running"); return; } running = true; scheduler = Executors.newSingleThreadScheduledExecutor(); scheduler.scheduleAtFixedRate(() -> { try (Connection conn = dataSource.getConnection(); PreparedStatement stmt = conn.prepareStatement(query); ResultSet rs = stmt.executeQuery()) { while (rs.next() && running) { try { T element = rowMapper.mapRow(rs); context.emit(element); elementsEmitted.incrementAndGet(); } catch (Exception e) { log.error("Failed to map row from JDBC source", e); errors.incrementAndGet(); context.markFailed("row-" + elementsEmitted.get(), e); } } } catch (Exception e) { log.error("Error in JDBC source", e); errors.incrementAndGet(); } }, 0, pollIntervalMs, TimeUnit.MILLISECONDS); log.info("JDBC source started with query: {}", query); } @Override public void stop() { running = false; if (scheduler != null) { scheduler.shutdown(); try { if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) { scheduler.shutdownNow(); } } catch (InterruptedException e) { scheduler.shutdownNow(); Thread.currentThread().interrupt(); } } } @Override public SourceStats getStats() { return new SourceStats( getName(), elementsEmitted.get(), errors.get(), elementsEmitted.get() / (System.currentTimeMillis() / 1000.0) ); } public interface RowMapper<T> { T mapRow(ResultSet rs) throws SQLException; } }

Stream Processing

3. Transform Operations

// Core transform implementations @Component @Slf4j public class MapTransform<T, R> implements Transform<T, R> { private final Function<T, R> mapper; private final String name; private final AtomicLong processedCount = new AtomicLong(0); private final AtomicLong errorCount = new AtomicLong(0); public MapTransform(String name, Function<T, R> mapper) { this.name = name; this.mapper = mapper; } @Override public String getName() { return name; } @Override public R apply(T input) { try { R result = mapper.apply(input); processedCount.incrementAndGet(); return result; } catch (Exception e) { errorCount.incrementAndGet(); onError(input, e); throw new TransformException("Mapping failed for input: " + input, e); } } @Override public void onError(T input, Throwable error) { log.error("Error in map transform {} for input: {}", name, input, error); } public long getProcessedCount() { return processedCount.get(); } public long getErrorCount() { return errorCount.get(); } } @Component @Slf4j public class FilterTransform<T> implements Transform<T, T> { private final Predicate<T> predicate; private final String name; private final AtomicLong processedCount = new AtomicLong(0); private final AtomicLong filteredCount = new AtomicLong(0); private final AtomicLong errorCount = new AtomicLong(0); public FilterTransform(String name, Predicate<T> predicate) { this.name = name; this.predicate = predicate; } @Override public String getName() { return name; } @Override public T apply(T input) { try { processedCount.incrementAndGet(); if (predicate.test(input)) { return input; } else { filteredCount.incrementAndGet(); return null; // Filtered out } } catch (Exception e) { errorCount.incrementAndGet(); onError(input, e); throw new TransformException("Filter failed for input: " + input, e); } } @Override public void onError(T input, Throwable error) { log.error("Error in filter transform {} for input: {}", name, input, error); } public long getProcessedCount() { return processedCount.get(); } public long getFilteredCount() { return filteredCount.get(); } public long getErrorCount() { return errorCount.get(); } } // Complex transform: Keyed operations @Component @Slf4j public class KeyedTransform<T, K, R> implements Transform<T, R> { private final Function<T, K> keyExtractor; private final Function<List<T>, R> aggregator; private final String name; private final int windowSize; private final Map<K, List<T>> keyedBuffer; private final AtomicLong processedCount = new AtomicLong(0); private final AtomicLong errorCount = new AtomicLong(0); public KeyedTransform(String name, Function<T, K> keyExtractor, Function<List<T>, R> aggregator, int windowSize) { this.name = name; this.keyExtractor = keyExtractor; this.aggregator = aggregator; this.windowSize = windowSize; this.keyedBuffer = new ConcurrentHashMap<>(); } @Override public String getName() { return name; } @Override public R apply(T input) { try { K key = keyExtractor.apply(input); List<T> buffer = keyedBuffer.computeIfAbsent(key, k -> new ArrayList<>()); synchronized (buffer) { buffer.add(input); if (buffer.size() >= windowSize) { R result = aggregator.apply(new ArrayList<>(buffer)); buffer.clear(); processedCount.incrementAndGet(); return result; } } processedCount.incrementAndGet(); return null; // Not enough elements for aggregation yet } catch (Exception e) { errorCount.incrementAndGet(); onError(input, e); throw new TransformException("Keyed transform failed for input: " + input, e); } } @Override public void onError(T input, Throwable error) { log.error("Error in keyed transform {} for input: {}", name, input, error); } // Flush remaining elements public List<R> flush() { List<R> results = new ArrayList<>(); for (List<T> buffer : keyedBuffer.values()) { synchronized (buffer) { if (!buffer.isEmpty()) { results.add(aggregator.apply(new ArrayList<>(buffer))); buffer.clear(); } } } return results; } } // Enrichment transform @Component @Slf4j public class EnrichmentTransform<T, R> implements Transform<T, R> { private final Function<T, R> enricher; private final String name; private final Cache<String, R> cache; private final AtomicLong processedCount = new AtomicLong(0); private final AtomicLong cacheHits = new AtomicLong(0); private final AtomicLong errorCount = new AtomicLong(0); public EnrichmentTransform(String name, Function<T, R> enricher, long cacheSize, Duration cacheTtl) { this.name = name; this.enricher = enricher; this.cache = Caffeine.newBuilder() .maximumSize(cacheSize) .expireAfterWrite(cacheTtl) .build(); } @Override public String getName() { return name; } @Override public R apply(T input) { try { String cacheKey = generateCacheKey(input); R result = cache.getIfPresent(cacheKey); if (result == null) { result = enricher.apply(input); cache.put(cacheKey, result); } else { cacheHits.incrementAndGet(); } processedCount.incrementAndGet(); return result; } catch (Exception e) { errorCount.incrementAndGet(); onError(input, e); throw new TransformException("Enrichment failed for input: " + input, e); } } @Override public void onError(T input, Throwable error) { log.error("Error in enrichment transform {} for input: {}", name, input, error); } private String generateCacheKey(T input) { return Objects.hash(input).toString(); } public double getCacheHitRate() { return processedCount.get() > 0 ? (double) cacheHits.get() / processedCount.get() : 0.0; } }

Batch Processing

4. Batch Pipeline Components

// Batch pipeline for processing large datasets @Component @Slf4j public class BatchPipeline<T> { private final String name; private final ExecutorService executor; private final PipelineMetrics metrics; private final int batchSize; private final Duration flushInterval; private Source<T> source; private final List<Transform<T, ?>> transforms; private Sink<T> sink; private volatile boolean running = false; private final List<T> batchBuffer; public BatchPipeline(String name, PipelineMetrics metrics, int batchSize, Duration flushInterval) { this.name = name; this.metrics = metrics; this.batchSize = batchSize; this.flushInterval = flushInterval; this.transforms = new CopyOnWriteArrayList<>(); this.batchBuffer = Collections.synchronizedList(new ArrayList<>()); this.executor = Executors.newFixedThreadPool(5, new ThreadFactoryBuilder() .setNameFormat("batch-pipeline-" + name + "-%d") .build()); } public BatchPipeline<T> from(Source<T> source) { this.source = source; return this; } @SuppressWarnings("unchecked") public <R> BatchPipeline<R> transform(Transform<T, R> transform) { this.transforms.add(transform); // In batch processing, we typically maintain the same pipeline return (BatchPipeline<R>) this; } public BatchPipeline<T> to(Sink<T> sink) { this.sink = sink; return this; } public void start() { if (running) { log.warn("Batch pipeline {} is already running", name); return; } running = true; // Start batch flusher executor.submit(this::batchFlusher); // Start source source.start(new SourceContext<T>() { @Override public void emit(T element) { addToBatch(element); } @Override public void markProcessed(String elementId) { metrics.recordElementProcessed(name, elementId); } @Override public void markFailed(String elementId, Throwable error) { metrics.recordElementFailed(name, elementId, error); } }); log.info("Batch pipeline {} started with batch size: {}, flush interval: {}", name, batchSize, flushInterval); } public void stop() { running = false; source.stop(); // Flush remaining elements flushBatch(); executor.shutdown(); try { if (!executor.awaitTermination(30, TimeUnit.SECONDS)) { executor.shutdownNow(); } } catch (InterruptedException e) { executor.shutdownNow(); Thread.currentThread().interrupt(); } metrics.recordPipelineStop(name); log.info("Batch pipeline {} stopped", name); } private void addToBatch(T element) { synchronized (batchBuffer) { batchBuffer.add(element); if (batchBuffer.size() >= batchSize) { flushBatch(); } } } private void batchFlusher() { while (running) { try { Thread.sleep(flushInterval.toMillis()); flushBatch(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { log.error("Error in batch flusher for pipeline {}", name, e); } } } @SuppressWarnings("unchecked") private void flushBatch() { List<T> currentBatch; synchronized (batchBuffer) { if (batchBuffer.isEmpty()) { return; } currentBatch = new ArrayList<>(batchBuffer); batchBuffer.clear(); } if (!currentBatch.isEmpty()) { executor.submit(() -> processBatch(currentBatch)); } } private void processBatch(List<T> batch) { try { List<T> currentBatch = batch; // Apply transforms for (Transform transform : transforms) { List<Object> transformedBatch = new ArrayList<>(); for (T element : currentBatch) { try { Object transformed = transform.apply(element); if (transformed != null) { transformedBatch.add(transformed); } } catch (Exception e) { log.error("Error applying transform {} to element", transform.getName(), e); metrics.recordElementFailed(name, element.toString(), e); } } currentBatch = (List<T>) transformedBatch; } // Write to sink if (sink != null && !currentBatch.isEmpty()) { sink.writeBatch(currentBatch); metrics.recordBatchProcessed(name, currentBatch.size()); } } catch (Exception e) { log.error("Error processing batch in pipeline {}", name, e); metrics.recordBatchFailed(name, batch.size(), e); } } } // Batch window for time-based batching @Component @Slf4j public class TimeWindow<T> implements WindowStrategy<T> { private final Duration windowSize; private final Duration windowSlide; private final List<T> windowBuffer; private final List<WindowProcessor<T>> processors; private ScheduledExecutorService windowScheduler; private volatile boolean running = false; public TimeWindow(Duration windowSize, Duration windowSlide) { this.windowSize = windowSize; this.windowSlide = windowSlide; this.windowBuffer = Collections.synchronizedList(new ArrayList<>()); this.processors = new CopyOnWriteArrayList<>(); } @Override public String getName() { return "time-window-" + windowSize.toMillis() + "ms"; } @Override public void process(T element) { if (running) { windowBuffer.add(element); } } public void addProcessor(WindowProcessor<T> processor) { processors.add(processor); } public void start() { if (running) { log.warn("Time window is already running"); return; } running = true; windowScheduler = Executors.newSingleThreadScheduledExecutor(); // Schedule window processing windowScheduler.scheduleAtFixedRate(this::processWindow, windowSlide.toMillis(), windowSlide.toMillis(), TimeUnit.MILLISECONDS); log.info("Time window started with size: {}, slide: {}", windowSize, windowSlide); } public void stop() { running = false; if (windowScheduler != null) { windowScheduler.shutdown(); try { if (!windowScheduler.awaitTermination(5, TimeUnit.SECONDS)) { windowScheduler.shutdownNow(); } } catch (InterruptedException e) { windowScheduler.shutdownNow(); Thread.currentThread().interrupt(); } } // Process remaining elements processWindow(); log.info("Time window stopped"); } private void processWindow() { List<T> currentWindow; synchronized (windowBuffer) { currentWindow = new ArrayList<>(windowBuffer); // For sliding windows, we might keep some elements // For tumbling windows, we clear the buffer if (windowSize.equals(windowSlide)) { windowBuffer.clear(); } else { // Keep elements from the last windowSize period long cutoffTime = System.currentTimeMillis() - windowSize.toMillis(); // This is simplified - in real implementation, you'd need timestamps windowBuffer.removeIf(element -> shouldRemoveFromWindow(element, cutoffTime)); } } if (!currentWindow.isEmpty()) { for (WindowProcessor<T> processor : processors) { try { processor.process(currentWindow); } catch (Exception e) { log.error("Error in window processor {}", processor.getName(), e); } } } } private boolean shouldRemoveFromWindow(T element, long cutoffTime) { // Implementation depends on how you track element timestamps // This is a placeholder implementation return true; // Remove all elements for simplicity } public interface WindowProcessor<T> { String getName(); void process(List<T> window); } }

Windowed Operations

5. Advanced Windowing Strategies

// Window manager for different window types @Component @Slf4j public class WindowManager<T> { private final Map<String, WindowStrategy<T>> windows; private final PipelineMetrics metrics; public WindowManager(PipelineMetrics metrics) { this.windows = new ConcurrentHashMap<>(); this.metrics = metrics; } public void registerWindow(String name, WindowStrategy<T> window) { windows.put(name, window); log.info("Registered window: {}", name); } public void processElement(String windowName, T element) { WindowStrategy<T> window = windows.get(windowName); if (window != null) { window.process(element); metrics.recordWindowElementProcessed(windowName); } else { log.warn("Window not found: {}", windowName); } } public void startAll() { windows.values().forEach(window -> { try { if (window instanceof ManageableWindow) { ((ManageableWindow) window).start(); } } catch (Exception e) { log.error("Failed to start window: {}", window.getName(), e); } }); } public void stopAll() { windows.values().forEach(window -> { try { if (window instanceof ManageableWindow) { ((ManageableWindow) window).stop(); } } catch (Exception e) { log.error("Failed to stop window: {}", window.getName(), e); } }); } public interface ManageableWindow { void start(); void stop(); } } // Session window for grouping events by session @Component @Slf4j public class SessionWindow<T> implements WindowStrategy<T>, WindowManager.ManageableWindow { private final Function<T, String> sessionIdExtractor; private final Function<T, Long> timestampExtractor; private final Duration sessionTimeout; private final Duration windowSlide; private final Map<String, Session<T>> activeSessions; private final List<WindowProcessor<T>> processors; private ScheduledExecutorSession cleanupScheduler; private volatile boolean running = false; public SessionWindow(Function<T, String> sessionIdExtractor, Function<T, Long> timestampExtractor, Duration sessionTimeout, Duration windowSlide) { this.sessionIdExtractor = sessionIdExtractor; this.timestampExtractor = timestampExtractor; this.sessionTimeout = sessionTimeout; this.windowSlide = windowSlide; this.activeSessions = new ConcurrentHashMap<>(); this.processors = new CopyOnWriteArrayList<>(); } @Override public String getName() { return "session-window-" + sessionTimeout.toMillis() + "ms"; } @Override public void process(T element) { if (!running) return; String sessionId = sessionIdExtractor.apply(element); long timestamp = timestampExtractor.apply(element); Session<T> session = activeSessions.computeIfAbsent(sessionId, id -> new Session<>(id, timestamp)); session.addElement(element, timestamp); session.updateLastActivity(timestamp); } @Override public void start() { if (running) { log.warn("Session window is already running"); return; } running = true; cleanupScheduler = Executors.newSingleThreadScheduledExecutor(); // Schedule session cleanup cleanupScheduler.scheduleAtFixedRate(this::cleanupExpiredSessions, windowSlide.toMillis(), windowSlide.toMillis(), TimeUnit.MILLISECONDS); log.info("Session window started with timeout: {}", sessionTimeout); } @Override public void stop() { running = false; if (cleanupScheduler != null) { cleanupScheduler.shutdown(); try { if (!cleanupScheduler.awaitTermination(5, TimeUnit.SECONDS)) { cleanupScheduler.shutdownNow(); } } catch (InterruptedException e) { cleanupScheduler.shutdownNow(); Thread.currentThread().interrupt(); } } // Process all remaining sessions closeAllSessions(); log.info("Session window stopped"); } public void addProcessor(WindowProcessor<T> processor) { processors.add(processor); } private void cleanupExpiredSessions() { long currentTime = System.currentTimeMillis(); long timeoutMillis = sessionTimeout.toMillis(); Iterator<Map.Entry<String, Session<T>>> iterator = activeSessions.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<String, Session<T>> entry = iterator.next(); Session<T> session = entry.getValue(); if (currentTime - session.getLastActivity() > timeoutMillis) { // Session expired, process and remove processSession(session); iterator.remove(); } } } private void closeAllSessions() { for (Session<T> session : activeSessions.values()) { processSession(session); } activeSessions.clear(); } private void processSession(Session<T> session) { List<T> sessionElements = session.getElements(); if (!sessionElements.isEmpty()) { for (WindowProcessor<T> processor : processors) { try { processor.process(sessionElements); } catch (Exception e) { log.error("Error processing session {} with processor {}", session.getId(), processor.getName(), e); } } } } private static class Session<T> { private final String id; private final List<T> elements; private long lastActivity; private final long startTime; public Session(String id, long startTime) { this.id = id; this.elements = new ArrayList<>(); this.startTime = startTime; this.lastActivity = startTime; } public void addElement(T element, long timestamp) { elements.add(element); } public void updateLastActivity(long timestamp) { this.lastActivity = timestamp; } public String getId() { return id; } public List<T> getElements() { return elements; } public long getLastActivity() { return lastActivity; } public long getStartTime() { return startTime; } } } // Global window for processing all elements as one group @Component @Slf4j public class GlobalWindow<T> implements WindowStrategy<T>, WindowManager.ManageableWindow { private final List<T> allElements; private final List<WindowProcessor<T>> processors; private final AtomicBoolean processed; public GlobalWindow() { this.allElements = Collections.synchronizedList(new ArrayList<>()); this.processors = new CopyOnWriteArrayList<>(); this.processed = new AtomicBoolean(false); } @Override public String getName() { return "global-window"; } @Override public void process(T element) { if (!processed.get()) { allElements.add(element); } } @Override public void start() { processed.set(false); log.info("Global window started"); } @Override public void stop() { if (processed.compareAndSet(false, true)) { processAllElements(); } log.info("Global window stopped"); } public void addProcessor(WindowProcessor<T> processor) { processors.add(processor); } private void processAllElements() { List<T> elements = new ArrayList<>(allElements); if (!elements.isEmpty()) { for (WindowProcessor<T> processor : processors) { try { processor.process(elements); } catch (Exception e) { log.error("Error in global window processor {}", processor.getName(), e); } } } allElements.clear(); } public void triggerProcessing() { stop(); start(); } }

State Management

6. Stateful Processing

// State management for pipeline operations @Component @Slf4j public class PipelineStateManager { private final StateStore stateStore; private final Map<String, PipelineState> pipelineStates; public PipelineStateManager(StateStore stateStore) { this.stateStore = stateStore; this.pipelineStates = new ConcurrentHashMap<>(); } public <T> PipelineState<T> getOrCreateState(String pipelineName, String stateName, Class<T> stateType, T initialState) { String fullStateName = pipelineName + ":" + stateName; @SuppressWarnings("unchecked") PipelineState<T> state = (PipelineState<T>) pipelineStates.get(fullStateName); if (state == null) { state = new PipelineState<>(fullStateName, stateType, initialState, stateStore); pipelineStates.put(fullStateName, state); } return state; } public void saveAllStates() { pipelineStates.values().forEach(PipelineState::save); } public void restoreAllStates() { pipelineStates.values().forEach(PipelineState::restore); } public void clearState(String pipelineName, String stateName) { String fullStateName = pipelineName + ":" + stateName; PipelineState<?> state = pipelineStates.remove(fullStateName); if (state != null) { state.clear(); } } } // Pipeline state with persistence @Component @Slf4j public class PipelineState<T> { private final String name; private final Class<T> type; private final StateStore stateStore; private T currentState; private final T initialState; public PipelineState(String name, Class<T> type, T initialState, StateStore stateStore) { this.name = name; this.type = type; this.initialState = initialState; this.stateStore = stateStore; this.currentState = initialState; restore(); } public T get() { return currentState; } public void update(Function<T, T> updater) { synchronized (this) { T newState = updater.apply(currentState); if (newState != null) { currentState = newState; } } } public void set(T newState) { synchronized (this) { currentState = newState; } } public void save() { try { stateStore.save(name, currentState); log.debug("Saved state for: {}", name); } catch (Exception e) { log.error("Failed to save state for: {}", name, e); } } public void restore() { try { T savedState = stateStore.load(name, type); if (savedState != null) { currentState = savedState; log.debug("Restored state for: {}", name); } } catch (Exception e) { log.warn("Failed to restore state for: {}, using initial state", name, e); currentState = initialState; } } public void clear() { synchronized (this) { currentState = initialState; } stateStore.delete(name); } public String getName() { return name; } } // State store interface with multiple implementations public interface StateStore { <T> void save(String key, T value); <T> T load(String key, Class<T> type); void delete(String key); boolean exists(String key); } // In-memory state store @Component @Slf4j public class InMemoryStateStore implements StateStore { private final Map<String, Object> storage; private final ObjectMapper objectMapper; public InMemoryStateStore(ObjectMapper objectMapper) { this.storage = new ConcurrentHashMap<>(); this.objectMapper = objectMapper; } @Override public <T> void save(String key, T value) { storage.put(key, value); } @Override @SuppressWarnings("unchecked") public <T> T load(String key, Class<T> type) { Object value = storage.get(key); if (value != null && type.isInstance(value)) { return (T) value; } return null; } @Override public void delete(String key) { storage.remove(key); } @Override public boolean exists(String key) { return storage.containsKey(key); } } // File-based state store @Component @Slf4j public class FileStateStore implements StateStore { private final Path storageDir; private final ObjectMapper objectMapper; public FileStateStore(@Value("${app.state.store.dir:/tmp/pipeline-state}") String storageDir, ObjectMapper objectMapper) throws IOException { this.storageDir = Paths.get(storageDir); this.objectMapper = objectMapper; if (!Files.exists(this.storageDir)) { Files.createDirectories(this.storageDir); } } @Override public <T> void save(String key, T value) { try { Path filePath = storageDir.resolve(key + ".json"); String json = objectMapper.writeValueAsString(value); Files.write(filePath, json.getBytes(), StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING); } catch (Exception e) { throw new StateStoreException("Failed to save state for key: " + key, e); } } @Override public <T> T load(String key, Class<T> type) { try { Path filePath = storageDir.resolve(key + ".json"); if (!Files.exists(filePath)) { return null; } String json = new String(Files.readAllBytes(filePath)); return objectMapper.readValue(json, type); } catch (Exception e) { throw new StateStoreException("Failed to load state for key: " + key, e); } } @Override public void delete(String key) { try { Path filePath = storageDir.resolve(key + ".json"); Files.deleteIfExists(filePath); } catch (Exception e) { throw new StateStoreException("Failed to delete state for key: " + key, e); } } @Override public boolean exists(String key) { Path filePath = storageDir.resolve(key + ".json"); return Files.exists(filePath); } } // Stateful transform example @Component @Slf4j public class StatefulAggregationTransform<T, K, R> implements Transform<T, R> { private final Function<T, K> keyExtractor; private final Function<T, Double> valueExtractor; private final AggregationFunction aggregationFunction; private final PipelineStateManager stateManager; private final String pipelineName; private final String transformName; public StatefulAggregationTransform(String pipelineName, String transformName, Function<T, K> keyExtractor, Function<T, Double> valueExtractor, AggregationFunction aggregationFunction, PipelineStateManager stateManager) { this.pipelineName = pipelineName; this.transformName = transformName; this.keyExtractor = keyExtractor; this.valueExtractor = valueExtractor; this.aggregationFunction = aggregationFunction; this.stateManager = stateManager; } @Override public String getName() { return transformName; } @Override @SuppressWarnings("unchecked") public R apply(T input) { K key = keyExtractor.apply(input); Double value = valueExtractor.apply(input); String stateName = "aggregation-" + key.toString(); PipelineState<AggregationState> state = stateManager.getOrCreateState( pipelineName, stateName, AggregationState.class, new AggregationState()); AggregationState currentState = state.get(); AggregationState newState = aggregationFunction.aggregate(currentState, value); state.set(newState); return (R) new AggregationResult(key, newState, System.currentTimeMillis()); } @Override public void onError(T input, Throwable error) { log.error("Error in stateful aggregation transform for input: {}", input, error); } public static class AggregationState { private double sum = 0.0; private long count = 0; private double min = Double.MAX_VALUE; private double max = Double.MIN_VALUE; // Getters and setters public double getSum() { return sum; } public void setSum(double sum) { this.sum = sum; } public long getCount() { return count; } public void setCount(long count) { this.count = count; } public double getMin() { return min; } public void setMin(double min) { this.min = min; } public double getMax() { return max; } public void setMax(double max) { this.max = max; } public double getAverage() { return count > 0 ? sum / count : 0.0; } } public interface AggregationFunction { AggregationState aggregate(AggregationState current, Double value); } public static class SumAggregation implements AggregationFunction { @Override public AggregationState aggregate(AggregationState current, Double value) { current.setSum(current.getSum() + value); current.setCount(current.getCount() + 1); current.setMin(Math.min(current.getMin(), value)); current.setMax(Math.max(current.getMax(), value)); return current; } } public record AggregationResult<K>(K key, AggregationState state, long timestamp) {} }

Fault Tolerance

7. Error Handling and Recovery

// Circuit breaker for pipeline components @Component @Slf4j public class PipelineCircuitBreaker { private final Map<String, CircuitBreaker> breakers; private final CircuitBreakerConfig defaultConfig; public PipelineCircuitBreaker() { this.breakers = new ConcurrentHashMap<>(); this.defaultConfig = CircuitBreakerConfig.custom() .failureRateThreshold(50) .slowCallRateThreshold(50) .waitDurationInOpenState(Duration.ofSeconds(30)) .slowCallDurationThreshold(Duration.ofSeconds(5)) .permittedNumberOfCallsInHalfOpenState(3) .minimumNumberOfCalls(10) .slidingWindowType(CircuitBreakerConfig.SlidingWindowType.COUNT_BASED) .slidingWindowSize(20) .build(); } public <T> T execute(String breakerName, Supplier<T> supplier) { CircuitBreaker breaker = breakers.computeIfAbsent(breakerName, name -> CircuitBreaker.of(name, defaultConfig)); return breaker.executeSupplier(supplier); } public void execute(String breakerName, Runnable runnable) { CircuitBreaker breaker = breakers.computeIfAbsent(breakerName, name -> CircuitBreaker.of(name, defaultConfig)); breaker.executeRunnable(runnable); } public CircuitBreaker.State getState(String breakerName) { CircuitBreaker breaker = breakers.get(breakerName); return breaker != null ? breaker.getState() : CircuitBreaker.State.CLOSED; } } // Retry mechanism with exponential backoff @Component @Slf4j public class PipelineRetryManager { private final Map<String, Retry> retries; private final RetryConfig defaultConfig; public PipelineRetryManager() { this.retries = new ConcurrentHashMap<>(); this.defaultConfig = RetryConfig.custom() .maxAttempts(3) .waitDuration(Duration.ofSeconds(1)) .retryExceptions(IOException.class, TimeoutException.class) .ignoreExceptions(NullPointerException.class) .build(); } public <T> T executeWithRetry(String retryName, Supplier<T> supplier) { Retry retry = retries.computeIfAbsent(retryName, name -> Retry.of(name, defaultConfig)); return retry.executeSupplier(supplier); } public void executeWithRetry(String retryName, Runnable runnable) { Retry retry = retries.computeIfAbsent(retryName, name -> Retry.of(name, defaultConfig)); retry.executeRunnable(runnable); } public RetryConfig getRetryConfig(String retryName) { Retry retry = retries.get(retryName); return retry != null ? retry.getRetryConfig() : defaultConfig; } } // Dead letter queue for failed elements @Component @Slf4j public class DeadLetterQueue<T> { private final Sink<T> dlqSink; private final ObjectMapper objectMapper; private final String queueName; public DeadLetterQueue(Sink<T> dlqSink, ObjectMapper objectMapper, String queueName) { this.dlqSink = dlqSink; this.objectMapper = objectMapper; this.queueName = queueName; } public void sendToDLQ(T element, Throwable error) { try { DLQMessage<T> message = new DLQMessage<>(element, error, System.currentTimeMillis()); dlqSink.write((T) message); // Cast is safe due to type erasure log.warn("Sent element to DLQ {}: {}", queueName, element); } catch (Exception e) { log.error("Failed to send element to DLQ {}: {}", queueName, element, e); } } public void sendToDLQ(T element, String errorMessage) { sendToDLQ(element, new RuntimeException(errorMessage)); } public static class DLQMessage<T> { private final T originalElement; private final String errorMessage; private final String stackTrace; private final long timestamp; private final String messageId; public DLQMessage(T originalElement, Throwable error, long timestamp) { this.originalElement = originalElement; this.errorMessage = error.getMessage(); this.stackTrace = getStackTrace(error); this.timestamp = timestamp; this.messageId = UUID.randomUUID().toString(); } private String getStackTrace(Throwable error) { StringWriter sw = new StringWriter(); error.printStackTrace(new PrintWriter(sw)); return sw.toString(); } // Getters public T getOriginalElement() { return originalElement; } public String getErrorMessage() { return errorMessage; } public String getStackTrace() { return stackTrace; } public long getTimestamp() { return timestamp; } public String getMessageId() { return messageId; } } } // Checkpointing for pipeline state @Component @Slf4j public class PipelineCheckpointer { private final StateStore stateStore; private final ScheduledExecutorService checkpointScheduler; private final Map<String, Checkpointable> checkpointables; public PipelineCheckpointer(StateStore stateStore) { this.stateStore = stateStore; this.checkpointScheduler = Executors.newSingleThreadScheduledExecutor(); this.checkpointables = new ConcurrentHashMap<>(); } public void registerCheckpointable(String name, Checkpointable checkpointable) { checkpointables.put(name, checkpointable); log.info("Registered checkpointable: {}", name); } public void startCheckpointing(Duration interval) { checkpointScheduler.scheduleAtFixedRate(this::performCheckpoint, interval.toMillis(), interval.toMillis(), TimeUnit.MILLISECONDS); log.info("Started checkpointing with interval: {}", interval); } public void stopCheckpointing() { checkpointScheduler.shutdown(); try { if (!checkpointScheduler.awaitTermination(5, TimeUnit.SECONDS)) { checkpointScheduler.shutdownNow(); } } catch (InterruptedException e) { checkpointScheduler.shutdownNow(); Thread.currentThread().interrupt(); } log.info("Stopped checkpointing"); } public void performCheckpoint() { checkpointables.forEach((name, checkpointable) -> { try { checkpointable.checkpoint(); log.debug("Performed checkpoint for: {}", name); } catch (Exception e) { log.error("Failed to perform checkpoint for: {}", name, e); } }); } public void restoreFromCheckpoint() { checkpointables.forEach((name, checkpointable) -> { try { checkpointable.restore(); log.debug("Restored from checkpoint for: {}", name); } catch (Exception e) { log.error("Failed to restore from checkpoint for: {}", name, e); } }); } public interface Checkpointable { void checkpoint(); void restore(); } }

Monitoring & Metrics

8. Comprehensive Monitoring

// Metrics service for pipeline monitoring @Component @Slf4j public class PipelineMetrics { private final MeterRegistry meterRegistry; private final Map<String, Counter> processedCounters; private final Map<String, Counter> failedCounters; private final Map<String, Timer> processingTimers; private final Map<String, Gauge> queueSizeGauges; public PipelineMetrics(MeterRegistry meterRegistry) { this.meterRegistry = meterRegistry; this.processedCounters = new ConcurrentHashMap<>(); this.failedCounters = new ConcurrentHashMap<>(); this.processingTimers = new ConcurrentHashMap<>(); this.queueSizeGauges = new ConcurrentHashMap<>(); } public void recordPipelineStart(String pipelineName) { meterRegistry.counter("pipeline.starts", "pipeline", pipelineName).increment(); } public void recordPipelineStop(String pipelineName) { meterRegistry.counter("pipeline.stops", "pipeline", pipelineName).increment(); } public void recordElementProcessed(String pipelineName, String elementId) { getProcessedCounter(pipelineName).increment(); meterRegistry.counter("pipeline.elements.processed", "pipeline", pipelineName).increment(); } public void recordElementFailed(String pipelineName, String elementId, Throwable error) { getFailedCounter(pipelineName).increment(); meterRegistry.counter("pipeline.elements.failed", "pipeline", pipelineName).increment(); meterRegistry.counter("pipeline.errors", "pipeline", pipelineName, "error", error.getClass().getSimpleName()).increment(); } public void recordBatchProcessed(String pipelineName, int batchSize) { meterRegistry.counter("pipeline.batches.processed", "pipeline", pipelineName).increment(); meterRegistry.summary("pipeline.batch.size", "pipeline", pipelineName).record(batchSize); } public void recordBatchFailed(String pipelineName, int batchSize, Throwable error) { meterRegistry.counter("pipeline.batches.failed", "pipeline", pipelineName).increment(); } public void recordWindowElementProcessed(String windowName) { meterRegistry.counter("window.elements.processed", "window", windowName).increment(); } public Timer.Sample startProcessingTimer(String pipelineName) { return Timer.start(meterRegistry); } public void stopProcessingTimer(Timer.Sample sample, String pipelineName) { sample.stop(getProcessingTimer(pipelineName)); } public void recordQueueSize(String queueName, int size) { Gauge gauge = queueSizeGauges.computeIfAbsent(queueName, name -> Gauge.builder("pipeline.queue.size") .tag("queue", name) .register(meterRegistry)); // Note: Gauge registration is handled by Micrometer, this is conceptual } public double getProcessedCount(String pipelineName) { Counter counter = processedCounters.get(pipelineName); return counter != null ? counter.count() : 0.0; } public double getFailedCount(String pipelineName) { Counter counter = failedCounters.get(pipelineName); return counter != null ? counter.count() : 0.0; } public double getProcessingRate(String pipelineName) { // This would need more sophisticated rate calculation return getProcessedCount(pipelineName) / (System.currentTimeMillis() / 1000.0); } private Counter getProcessedCounter(String pipelineName) { return processedCounters.computeIfAbsent(pipelineName, name -> meterRegistry.counter("pipeline.processed", "pipeline", name)); } private Counter getFailedCounter(String pipelineName) { return failedCounters.computeIfAbsent(pipelineName, name -> meterRegistry.counter("pipeline.failed", "pipeline", name)); } private Timer getProcessingTimer(String pipelineName) { return processingTimers.computeIfAbsent(pipelineName, name -> Timer.builder("pipeline.processing.time") .tag("pipeline", name) .register(meterRegistry)); } } // Health indicators for pipeline components @Component public class PipelineHealthIndicator implements HealthIndicator { private final PipelineMetrics metrics; private final Map<String, Pipeline<?>> pipelines; public PipelineHealthIndicator(PipelineMetrics metrics) { this.metrics = metrics; this.pipelines = new ConcurrentHashMap<>(); } public void registerPipeline(String name, Pipeline<?> pipeline) { pipelines.put(name, pipeline); } @Override public Health health() { Map<String, Object> details = new HashMap<>(); Health.Builder status = Health.up(); for (Map.Entry<String, Pipeline<?>> entry : pipelines.entrySet()) { String pipelineName = entry.getKey(); Pipeline<?> pipeline = entry.getValue(); PipelineStats stats = pipeline.getStats(); Map<String, Object> pipelineDetails = new HashMap<>(); pipelineDetails.put("processed", stats.processedCount()); pipelineDetails.put("failed", stats.failedCount()); pipelineDetails.put("rate", stats.processingRate()); // Check for stalled pipeline if (stats.processingRate() < 1.0) { // Less than 1 element per second pipelineDetails.put("status", "SLOW"); status = Health.down(); } else { pipelineDetails.put("status", "HEALTHY"); } details.put(pipelineName, pipelineDetails); } return status.withDetails(details).build(); } } // Pipeline dashboard service @Component @Slf4j public class PipelineDashboard { private final PipelineMetrics metrics; private final Map<String, Pipeline<?>> pipelines; private final ScheduledExecutorService dashboardScheduler; public PipelineDashboard(PipelineMetrics metrics) { this.metrics = metrics; this.pipelines = new ConcurrentHashMap<>(); this.dashboardScheduler = Executors.newSingleThreadScheduledExecutor(); } public void registerPipeline(String name, Pipeline<?> pipeline) { pipelines.put(name, pipeline); } public void startDashboardReporting(Duration interval) { dashboardScheduler.scheduleAtFixedRate(this::reportPipelineStats, interval.toMillis(), interval.toMillis(), TimeUnit.MILLISECONDS); } public void stopDashboardReporting() { dashboardScheduler.shutdown(); } private void reportPipelineStats() { log.info("=== Pipeline Dashboard Report ==="); for (Map.Entry<String, Pipeline<?>> entry : pipelines.entrySet()) { String pipelineName = entry.getKey(); Pipeline<?> pipeline = entry.getValue(); PipelineStats stats = pipeline.getStats(); log.info("Pipeline: {}", pipelineName); log.info(" Processed: {}, Failed: {}, Rate: {}/s", stats.processedCount(), stats.failedCount(), stats.processingRate()); if (stats.sourceStats() != null) { log.info(" Source: {} (emitted: {}, errors: {})", stats.sourceStats().sourceName(), stats.sourceStats().elementsEmitted(), stats.sourceStats().errors()); } if (stats.sinkStats() != null) { log.info(" Sink: {} (written: {}, errors: {})", stats.sinkStats().sinkName(), stats.sinkStats().elementsWritten(), stats.sinkStats().errors()); } } log.info("================================="); } public Map<String, Object> getPipelineMetrics(String pipelineName) { Pipeline<?> pipeline = pipelines.get(pipelineName); if (pipeline == null) { return Map.of("error", "Pipeline not found: " + pipelineName); } PipelineStats stats = pipeline.getStats(); return Map.of( "name", pipelineName, "processed", stats.processedCount(), "failed", stats.failedCount(), "rate", stats.processingRate(), "source", stats.sourceStats() != null ? Map.of( "name", stats.sourceStats().sourceName(), "emitted", stats.sourceStats().elementsEmitted(), "errors", stats.sourceStats().errors() ) : null, "sink", stats.sinkStats() != null ? Map.of( "name", stats.sinkStats().sinkName(), "written", stats.sinkStats().elementsWritten(), "errors", stats.sinkStats().errors() ) : null ); } }

Spring Boot Integration

9. Complete Spring Boot Configuration

# application.yml spring: application: name: dataflow-pipeline kafka: bootstrap-servers: localhost:9092 consumer: group-id: pipeline-consumer auto-offset-reset: earliest producer: acks: all retries: 3 datasource: url: jdbc:postgresql://localhost:5432/pipeline username: pipeline password: pipeline hikari: maximum-pool-size: 10 jackson: property-naming-strategy: SNAKE_CASE default-property-inclusion: NON_NULL # Pipeline Configuration pipeline: batch: size: 1000 flush-interval: 5s window: size: 1m slide: 30s state: store: file directory: /tmp/pipeline-state retry: max-attempts: 3 backoff-initial-interval: 1s circuit-breaker: failure-threshold: 50 wait-duration: 30s # Pipeline definitions pipelines: user-activity: source: type: kafka topic: user-events transforms: - type: filter condition: "eventType == 'CLICK'" - type: map field: "userId" - type: window strategy: session timeout: 30m sink: type: kafka topic: user-sessions order-processing: source: type: jdbc query: "SELECT * FROM orders WHERE status = 'PENDING'" poll-interval: 10s transforms: - type: enrich service: customer-service field: "customerId" - type: batch size: 100 sink: type: jdbc table: processed_orders # Management management: endpoints: web: exposure: include: health,metrics,info,prometheus endpoint: health: show-details: always metrics: export: prometheus: enabled: true distribution: percentiles-histogram: "[pipeline.processing.time]": true

10. Spring Configuration and Main Application

@Configuration @EnableConfigurationProperties(PipelineProperties.class) public class PipelineConfiguration { @Bean public PipelineMetrics pipelineMetrics(MeterRegistry meterRegistry) { return new PipelineMetrics(meterRegistry); } @Bean public PipelineStateManager pipelineStateManager(StateStore stateStore) { return new PipelineStateManager(stateStore); } @Bean @ConditionalOnProperty(name = "pipeline.state.store", havingValue = "file") public StateStore fileStateStore(PipelineProperties properties, ObjectMapper objectMapper) throws IOException { return new FileStateStore(properties.getState().getDirectory(), objectMapper); } @Bean @ConditionalOnMissingBean(StateStore.class) public StateStore inMemoryStateStore(ObjectMapper objectMapper) { return new InMemoryStateStore(objectMapper); } @Bean public PipelineCircuitBreaker pipelineCircuitBreaker() { return new PipelineCircuitBreaker(); } @Bean public PipelineRetryManager pipelineRetryManager() { return new PipelineRetryManager(); } @Bean public PipelineCheckpointer pipelineCheckpointer(StateStore stateStore) { return new PipelineCheckpointer(stateStore); } @Bean public PipelineHealthIndicator pipelineHealthIndicator(PipelineMetrics metrics) { return new PipelineHealthIndicator(metrics); } @Bean public PipelineDashboard pipelineDashboard(PipelineMetrics metrics) { return new PipelineDashboard(metrics); } @Bean public ObjectMapper objectMapper() { return new ObjectMapper() .registerModule(new JavaTimeModule()) .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS) .setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE); } } @ConfigurationProperties(prefix = "pipeline") @Data public class PipelineProperties { private Batch batch = new Batch(); private Window window = new Window(); private State state = new State(); private Retry retry = new Retry(); private CircuitBreaker circuitBreaker = new CircuitBreaker(); private Map<String, PipelineConfig> pipelines = new HashMap<>(); @Data public static class Batch { private int size = 1000; private Duration flushInterval = Duration.ofSeconds(5); } @Data public static class Window { private Duration size = Duration.ofMinutes(1); private Duration slide = Duration.ofSeconds(30); } @Data public static class State { private String store = "memory"; private String directory = "/tmp/pipeline-state"; } @Data public static class Retry { private int maxAttempts = 3; private Duration backoffInitialInterval = Duration.ofSeconds(1); } @Data public static class CircuitBreaker { private int failureThreshold = 50; private Duration waitDuration = Duration.ofSeconds(30); } @Data public static class PipelineConfig { private SourceConfig source; private List<TransformConfig> transforms = new ArrayList<>(); private SinkConfig sink; } @Data public static class SourceConfig { private String type; private String topic; private String query; private Duration pollInterval; } @Data public static class TransformConfig { private String type; private String condition; private String field; private String service; } @Data public static class SinkConfig { private String type; private String topic; private String table; } } // Main application with pipeline orchestration @SpringBootApplication @Slf4j public class DataflowPipelineApplication { public static void main(String[] args) { SpringApplication.run(DataflowPipelineApplication.class, args); } @Bean public CommandLineRunner pipelineOrchestrator(PipelineProperties properties, PipelineMetrics metrics, KafkaTemplate<String, String> kafkaTemplate, DataSource dataSource, ObjectMapper objectMapper) { return args -> { log.info("Initializing dataflow pipelines..."); for (Map.Entry<String, PipelineProperties.PipelineConfig> entry : properties.getPipelines().entrySet()) { String pipelineName = entry.getKey(); PipelineProperties.PipelineConfig config = entry.getValue(); createPipeline(pipelineName, config, metrics, kafkaTemplate, dataSource, objectMapper); } log.info("All pipelines initialized successfully"); }; } private void createPipeline(String name, PipelineProperties.PipelineConfig config, PipelineMetrics metrics, KafkaTemplate<String, String> kafkaTemplate, DataSource dataSource, ObjectMapper objectMapper) { // Implementation would create pipeline based on configuration log.info("Creating pipeline: {} with config: {}", name, config); // Example pipeline creation logic if ("user-activity".equals(name)) { createUserActivityPipeline(name, config, metrics, kafkaTemplate, objectMapper); } else if ("order-processing".equals(name)) { createOrderProcessingPipeline(name, config, metrics, dataSource, objectMapper); } } private void createUserActivityPipeline(String name, PipelineProperties.PipelineConfig config, PipelineMetrics metrics, KafkaTemplate<String, String> kafkaTemplate, ObjectMapper objectMapper) { // Create user activity pipeline Pipeline<String> pipeline = new DataflowPipeline<>(name, metrics); // Configure source KafkaSource<String> source = new KafkaSource<>(kafkaTemplate, objectMapper, String.class, config.getSource().getTopic(), "user-activity-group"); // Configure transforms pipeline.from(source) .filter(new FilterTransform<>("click-filter", event -> event.contains("\"eventType\":\"CLICK\""))) .transform(new MapTransform<>("extract-userId", event -> extractUserId(event))) .window(new SessionWindow<>( this::extractSessionId, event -> System.currentTimeMillis(), Duration.ofMinutes(30), Duration.ofSeconds(30) )); // Configure sink KafkaSink<String> sink = new KafkaSink<>(kafkaTemplate, config.getSink().getTopic(), objectMapper); pipeline.to(sink); pipeline.start(); log.info("Started user activity pipeline: {}", name); } private void createOrderProcessingPipeline(String name, PipelineProperties.PipelineConfig config, PipelineMetrics metrics, DataSource dataSource, ObjectMapper objectMapper) { // Create order processing pipeline BatchPipeline<String> pipeline = new BatchPipeline<>(name, metrics, properties.getBatch().getSize(), properties.getBatch().getFlushInterval()); // Configure source JdbcSource<String> source = new JdbcSource<>(dataSource, config.getSource().getQuery(), rs -> rs.getString("order_data"), config.getSource().getPollInterval().toMillis()); // Configure pipeline pipeline.from(source) .to(new JdbcSink<>(dataSource, config.getSink().getTable())); pipeline.start(); log.info("Started order processing pipeline: {}", name); } private String extractUserId(String event) { // Simplified implementation try { JsonNode node = objectMapper.readTree(event); return node.get("userId").asText(); } catch (Exception e) { return "unknown"; } } private String extractSessionId(String event) { // Simplified implementation try { JsonNode node = objectMapper.readTree(event); return node.get("sessionId").asText(); } catch (Exception e) { return "unknown-session"; } } } // REST controller for pipeline management @RestController @RequestMapping("/api/pipelines") @Slf4j public class PipelineController { private final PipelineDashboard dashboard; private final Map<String, Pipeline<?>> pipelines; public PipelineController(PipelineDashboard dashboard) { this.dashboard = dashboard; this.pipelines = new ConcurrentHashMap<>(); } @PostMapping("/{name}/start") public ResponseEntity<String> startPipeline(@PathVariable String name) { Pipeline<?> pipeline = pipelines.get(name); if (pipeline == null) { return ResponseEntity.notFound().build(); } pipeline.start(); return ResponseEntity.ok("Pipeline " + name + " started"); } @PostMapping("/{name}/stop") public ResponseEntity<String> stopPipeline(@PathVariable String name) { Pipeline<?> pipeline = pipelines.get(name); if (pipeline == null) { return ResponseEntity.notFound().build(); } pipeline.stop(); return ResponseEntity.ok("Pipeline " + name + " stopped"); } @GetMapping("/{name}/metrics") public ResponseEntity<Map<String, Object>> getPipelineMetrics(@PathVariable String name) { Map<String, Object> metrics = dashboard.getPipelineMetrics(name); if (metrics.containsKey("error")) { return ResponseEntity.notFound().build(); } return ResponseEntity.ok(metrics); } @GetMapping("/{name}/stats") public ResponseEntity<PipelineStats> getPipelineStats(@PathVariable String name) { Pipeline<?> pipeline = pipelines.get(name); if (pipeline == null) { return ResponseEntity.notFound().build(); } return ResponseEntity.ok(pipeline.getStats()); } public void registerPipeline(String name, Pipeline<?> pipeline) { pipelines.put(name, pipeline); dashboard.registerPipeline(name, pipeline); } }

This comprehensive dataflow pipeline implementation provides:

Modular pipeline architecture with sources, transforms, and sinks
Both stream and batch processing capabilities
Advanced windowing strategies for time-based operations
Robust state management with persistence
Comprehensive fault tolerance with circuit breakers and retries
Detailed monitoring and metrics for observability
Spring Boot integration for easy deployment and configuration

The pipeline is designed to be extensible, scalable, and production-ready for various data processing scenarios.

Table of Contents