Streaming

Testing Apache Flink and Spark Streaming Applications

09 Oct 2026 — 6 min read

Stream processing jobs are notoriously difficult to test. Flink and Spark Streaming jobs process unbounded data with complex windowing logic, watermarks, and stateful operations. The standard "run it and see" approach doesn't work — you need deterministic, reproducible tests that verify your transformation logic without running a full cluster.

This guide covers testing strategies for Flink and Spark Streaming that catch real bugs.

What to Test in Stream Processing

Stream processing bugs fall into three categories:

Transformation bugs: The map/filter/aggregate logic produces wrong results
Windowing bugs: Events fall into wrong windows, windows close too early/late, late events are dropped when they shouldn't be
State management bugs: State grows unbounded, state is not cleared correctly, state is inconsistent after recovery

Each requires different testing approaches.

Testing Apache Flink

Unit Testing Flink Functions

The fastest Flink tests don't touch the runtime at all:

// OrderAmountMapper.java
public class OrderAmountMapper extends RichMapFunction<OrderEvent, Double> {
    @Override
    public Double map(OrderEvent event) {
        if (event.getTotal() == null || event.getTotal().compareTo(BigDecimal.ZERO) < 0) {
            throw new IllegalArgumentException("Invalid order total: " + event.getTotal());
        }
        return event.getTotal().doubleValue();
    }
}

// OrderAmountMapperTest.java
class OrderAmountMapperTest {
    private OrderAmountMapper mapper = new OrderAmountMapper();
    
    @Test
    void map_convertsOrderTotalToDouble() throws Exception {
        OrderEvent event = OrderEvent.builder()
            .orderId(1L)
            .total(new BigDecimal("99.99"))
            .build();
        
        Double result = mapper.map(event);
        
        assertThat(result).isEqualTo(99.99);
    }
    
    @Test
    void map_throwsOnNullTotal() {
        OrderEvent event = OrderEvent.builder().orderId(1L).total(null).build();
        
        assertThatThrownBy(() -> mapper.map(event))
            .isInstanceOf(IllegalArgumentException.class)
            .hasMessageContaining("Invalid order total");
    }
    
    @Test
    void map_throwsOnNegativeTotal() {
        OrderEvent event = OrderEvent.builder()
            .orderId(1L)
            .total(new BigDecimal("-10.00"))
            .build();
        
        assertThatThrownBy(() -> mapper.map(event))
            .isInstanceOf(IllegalArgumentException.class);
    }
}

Testing with Flink's MiniCluster

For testing the full pipeline including windowing and state, use MiniClusterWithClientResource:

@ExtendWith(MiniClusterExtension.class)
class OrderRevenueJobTest {
    
    @Test
    void computesRevenuePer5MinuteWindow() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        // Use a manual time source for deterministic windowing
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        
        // Create test events with explicit timestamps
        List<OrderEvent> events = List.of(
            OrderEvent.of(1L, "US", new BigDecimal("100.00"), toEpoch("2024-01-01T10:00:00")),
            OrderEvent.of(2L, "US", new BigDecimal("200.00"), toEpoch("2024-01-01T10:02:00")),
            OrderEvent.of(3L, "EU", new BigDecimal("150.00"), toEpoch("2024-01-01T10:03:00")),
            // This one is in the next window
            OrderEvent.of(4L, "US", new BigDecimal("300.00"), toEpoch("2024-01-01T10:06:00")),
        );
        
        DataStream<OrderEvent> input = env.fromCollection(events)
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                    .withTimestampAssigner((e, t) -> e.getTimestamp())
            );
        
        // The job under test
        DataStream<RevenueResult> result = OrderRevenueJob.transform(input);
        
        // Collect results
        List<RevenueResult> collected = new ArrayList<>();
        result.addSink(new SinkFunction<>() {
            @Override
            public void invoke(RevenueResult value, Context ctx) {
                collected.add(value);
            }
        });
        
        env.execute("test");
        
        // First window (10:00-10:05): US=300, EU=150
        Map<String, Double> firstWindow = collected.stream()
            .filter(r -> r.getWindowEnd() == toEpoch("2024-01-01T10:05:00"))
            .collect(Collectors.toMap(RevenueResult::getRegion, RevenueResult::getRevenue));
        
        assertThat(firstWindow.get("US")).isEqualTo(300.0);
        assertThat(firstWindow.get("EU")).isEqualTo(150.0);
        
        // Second window (10:05-10:10): US=300
        Map<String, Double> secondWindow = collected.stream()
            .filter(r -> r.getWindowEnd() == toEpoch("2024-01-01T10:10:00"))
            .collect(Collectors.toMap(RevenueResult::getRegion, RevenueResult::getRevenue));
        
        assertThat(secondWindow.get("US")).isEqualTo(300.0);
        assertThat(secondWindow).doesNotContainKey("EU");
    }
}

Testing Watermark and Late Event Handling

Watermarks determine when windows close and how late events are handled. Test this explicitly:

@Test
void lateEventsWithinAllowedLatenessAreIncluded() throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
    // Allow 60 seconds of lateness
    TumblingEventTimeWindows window = TumblingEventTimeWindows.of(Time.minutes(5));
    
    List<OrderEvent> events = List.of(
        OrderEvent.of(1L, "US", new BigDecimal("100.00"), toEpoch("10:00")),
        OrderEvent.of(2L, "US", new BigDecimal("200.00"), toEpoch("10:04")),
        // Watermark advances to 10:10 (after next window events)
        OrderEvent.of(3L, "US", new BigDecimal("50.00"),  toEpoch("10:08")),  // advances watermark
        // Late event — arrives after window closed but within allowed lateness
        OrderEvent.of(4L, "US", new BigDecimal("25.00"),  toEpoch("10:03")),  // late!
    );
    
    // ... setup and execute
    
    // First window result should include the late event (25.00)
    double usRevenue = results.stream()
        .filter(r -> r.getWindowEnd() == toEpoch("10:05") && r.getRegion().equals("US"))
        .mapToDouble(RevenueResult::getRevenue)
        .sum();
    
    assertThat(usRevenue).isEqualTo(325.0, within(0.01));  // 100 + 200 + 25
}

@Test
void lateEventsBeyondAllowedLatenessAreDropped() throws Exception {
    // Similar setup but with event too late for allowed lateness window
    // Assert the very-late event is NOT in any result
}

Testing Apache Spark Streaming

Unit Testing Spark Transformations

Spark transformations can be tested without a cluster using local mode:

import pytest
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType
from datetime import datetime

@pytest.fixture(scope="session")
def spark():
    return SparkSession.builder \
        .master("local[2]") \
        .appName("test") \
        .config("spark.sql.shuffle.partitions", "2") \
        .getOrCreate()

@pytest.fixture(scope="session")
def order_schema():
    return StructType([
        StructField("order_id", LongType()),
        StructField("customer_id", LongType()),
        StructField("region", StringType()),
        StructField("total", DoubleType()),
        StructField("status", StringType()),
        StructField("event_time", LongType()),  # epoch millis
    ])


def test_revenue_aggregation(spark, order_schema):
    """Revenue aggregation sums only completed orders per region."""
    from your_app.transformations import compute_revenue_by_region
    
    data = [
        (1, 100, "US", 100.0, "completed", 1704067200000),
        (2, 101, "US", 200.0, "completed", 1704067260000),
        (3, 102, "EU", 150.0, "completed", 1704067200000),
        (4, 103, "US", 300.0, "cancelled", 1704067200000),  # excluded
    ]
    
    df = spark.createDataFrame(data, schema=order_schema)
    result = compute_revenue_by_region(df)
    
    rows = {row["region"]: row["revenue"] for row in result.collect()}
    
    assert rows["US"] == 300.0, f"US revenue wrong: {rows.get('US')}"
    assert rows["EU"] == 150.0, f"EU revenue wrong: {rows.get('EU')}"
    assert "US" in rows and rows["US"] != 600.0, "Cancelled orders should be excluded"


def test_deduplication_by_order_id(spark, order_schema):
    """Duplicate events for the same order are deduplicated."""
    from your_app.transformations import deduplicate_orders
    
    # Same order_id, different timestamps (duplicate events)
    data = [
        (1, 100, "US", 100.0, "completed", 1704067200000),
        (1, 100, "US", 100.0, "completed", 1704067260000),  # duplicate
        (2, 101, "US", 200.0, "completed", 1704067200000),
    ]
    
    df = spark.createDataFrame(data, schema=order_schema)
    result = deduplicate_orders(df)
    
    assert result.count() == 2, "Deduplication should produce exactly 2 distinct orders"
    
    order_ids = {row["order_id"] for row in result.select("order_id").collect()}
    assert order_ids == {1, 2}

Testing Structured Streaming with Memory Sink

def test_windowed_revenue_streaming(spark):
    """Windowed revenue aggregation produces correct results per time window."""
    from pyspark.sql.functions import window, sum as spark_sum, col
    from pyspark.sql.types import TimestampType
    
    # Create a mock streaming source using rate source
    # In real tests, use MemoryStream for precise control
    
    schema = StructType([
        StructField("order_id", LongType()),
        StructField("region", StringType()),
        StructField("total", DoubleType()),
        StructField("event_time", TimestampType()),
    ])
    
    # Use MemoryStream for deterministic test data
    from pyspark.sql.streaming import DataStreamWriter
    
    # Create test data as static DataFrame and treat as streaming
    test_data = [
        (1, "US", 100.0, datetime(2024, 1, 1, 10, 0, 0)),
        (2, "US", 200.0, datetime(2024, 1, 1, 10, 2, 0)),
        (3, "EU", 150.0, datetime(2024, 1, 1, 10, 3, 0)),
    ]
    
    df = spark.createDataFrame(test_data, schema=schema)
    
    result = df \
        .withWatermark("event_time", "1 minute") \
        .groupBy(
            window(col("event_time"), "5 minutes"),
            col("region")
        ) \
        .agg(spark_sum("total").alias("revenue"))
    
    rows = result.collect()
    revenue_by_region = {r["region"]: r["revenue"] for r in rows}
    
    assert revenue_by_region["US"] == 300.0
    assert revenue_by_region["EU"] == 150.0

Testing State Management

Stateful operations are the hardest to test because bugs may only appear after many events:

@Test
void sessionWindowClosesAfterInactivity() throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    
    // Session window closes after 30 minutes of inactivity
    List<ClickEvent> events = List.of(
        ClickEvent.of("user:1", "page_view", toEpoch("10:00")),
        ClickEvent.of("user:1", "page_view", toEpoch("10:15")),  // same session (< 30 min gap)
        ClickEvent.of("user:1", "page_view", toEpoch("10:29")),  // same session
        ClickEvent.of("user:1", "page_view", toEpoch("11:05")),  // NEW session (> 30 min gap)
    );
    
    // ... execute
    
    // Should produce 2 sessions for user:1
    long sessionCount = results.stream()
        .filter(r -> r.getUserId().equals("user:1"))
        .count();
    
    assertThat(sessionCount).isEqualTo(2);
    
    // First session: 3 events
    SessionResult firstSession = results.stream()
        .filter(r -> r.getUserId().equals("user:1"))
        .min(Comparator.comparing(SessionResult::getStartTime))
        .orElseThrow();
    
    assertThat(firstSession.getEventCount()).isEqualTo(3);
}

CI/CD Integration

# .github/workflows/streaming-tests.yml
name: Streaming Tests

on: [push, pull_request]

jobs:
  flink-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-java@v4
        with:
          java-version: '17'
          distribution: 'temurin'
      - name: Run Flink tests
        run: mvn test -pl streaming-jobs -Dtest="**/*Test" -DfailIfNoTests=false

  spark-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install pyspark pytest
      - name: Run Spark tests
        run: pytest tests/spark/ -v
        env:
          PYSPARK_PYTHON: python3

Common Stream Processing Testing Mistakes

Testing only with static data: Stream processing bugs often involve ordering and timing. Use event-time sources in tests, not just batch data.

Not testing watermarks: Watermark bugs cause late events to be dropped or windows to never close. Test watermark behavior explicitly with events at window boundaries.

Assuming deterministic ordering: Parallel stream processing doesn't guarantee ordering across partitions. Test that your results don't depend on event ordering.

No state cleanup tests: Stateful operations that don't clean up state cause memory leaks. Test that state is bounded even with long-running streams.

Too coarse-grained tests: "The job produces the right final output" hides where the bug is. Test individual transformations as units.

Stream processing testing requires two levels: unit tests for transformation logic (fast, no cluster) and integration tests with MiniCluster/local Spark for windowing and state behavior. The patterns here give you coverage of the hardest streaming bugs — watermark edge cases, late event handling, and state management — without requiring a production cluster.