11.1 Interactive Queries & Testing

Module Overview

Goal: Master interactive queries and testing of Kafka Streams applications.

What You'll Learn:

• Interactive queries (IQ) for querying state stores
• Key-value queries
• Range queries
• Windowed queries
• Distributed queries across instances
• TopologyTestDriver for unit testing
• Integration testing strategies

What's NOT Covered Here:

• State store internals (covered in Module 6.1)
• Windowing concepts (covered in Module 5.2)
• Monitoring metrics (covered in Module 12.1)

Key Insight: Interactive queries expose the state of your Streams application as a queryable database — testing ensures correctness before production.

1. What are Interactive Queries?

Definition

Interactive Queries (IQ) = querying state stores in real-time

Purpose:
├─ Expose aggregated data (counts, sums, etc.)
├─ Serve as read-only query layer
├─ Avoid external database for simple lookups
└─ Low-latency reads from local state

NOT for:
├─ Writing to state (read-only)
├─ Ad-hoc analytics (use ksqlDB or external DB)
└─ Complex joins (limited query capabilities)

Use Cases

1. Real-time dashboards
   └─ Query aggregated metrics (counts, sums)

2. Enrichment lookups
   └─ Query KTable state for enrichment

3. Materialized views
   └─ Query processed/aggregated data

4. Health checks
   └─ Verify application state

2. Enabling Interactive Queries

Materialized State Store

// State store MUST be materialized for queries
KTable<String, Long> counts = stream
    .groupByKey()
    .count(Materialized.as("user-counts"));  // Named store

// Without Materialized.as(), store not queryable

Store Name

Store name = identifier for queries

Rules:
├─ Must be unique per topology
├─ Use descriptive names
└─ Used to locate store across instances

Example:
└─ "user-counts", "windowed-aggregates", "customer-profiles"

3. Querying Local State

Get Store Instance

KafkaStreams streams = new KafkaStreams(topology, props);
streams.start();

// Wait for streams to be ready
while (streams.state() != State.RUNNING) {
    Thread.sleep(100);
}

// Get read-only state store
ReadOnlyKeyValueStore<String, Long> store = 
    streams.store(StoreQueryParameters.fromNameAndType(
        "user-counts",                    // Store name
        QueryableStoreTypes.keyValueStore()  // Store type
    ));

Key-Value Query

// Query by key
Long count = store.get("user123");

if (count != null) {
    System.out.println("User123 count: " + count);
} else {
    System.out.println("User123 not found");
}

Range Query

// Query all keys
KeyValueIterator<String, Long> all = store.all();

while (all.hasNext()) {
    KeyValue<String, Long> kv = all.next();
    System.out.println(kv.key + " -> " + kv.value);
}
all.close();  // Always close iterator

// Query range of keys
KeyValueIterator<String, Long> range = store.range("user100", "user200");

while (range.hasNext()) {
    KeyValue<String, Long> kv = range.next();
    System.out.println(kv.key + " -> " + kv.value);
}
range.close();

Approximate Count

// Get approximate number of entries
long approxCount = store.approximateNumEntries();
System.out.println("Approximate entries: " + approxCount);

4. Querying Windowed State

Windowed Store

// Create windowed aggregation
TimeWindowedKStream<String, String> windowed = stream
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)));

KTable<Windowed<String>, Long> counts = windowed.count(
    Materialized.as("windowed-counts")
);

Get Windowed Store

ReadOnlyWindowStore<String, Long> windowStore = 
    streams.store(StoreQueryParameters.fromNameAndType(
        "windowed-counts",
        QueryableStoreTypes.windowStore()
    ));

Query Specific Window

// Query specific key and time range
Instant now = Instant.now();
Instant from = now.minus(Duration.ofMinutes(10));
Instant to = now;

WindowStoreIterator<Long> results = windowStore.fetch(
    "user123",  // Key
    from,       // From time
    to          // To time
);

while (results.hasNext()) {
    KeyValue<Long, Long> kv = results.next();
    Instant windowStart = Instant.ofEpochMilli(kv.key);
    Long count = kv.value;
    System.out.println("Window " + windowStart + ": " + count);
}
results.close();

Query All Windows for Key

// Fetch all windows for a key (not time-bounded)
WindowStoreIterator<Long> allWindows = windowStore.fetch(
    "user123",
    Instant.EPOCH,
    Instant.now()
);

while (allWindows.hasNext()) {
    KeyValue<Long, Long> kv = allWindows.next();
    System.out.println("Timestamp: " + kv.key + ", Count: " + kv.value);
}
allWindows.close();

Query Multiple Keys

// Fetch all windows for key range
KeyValueIterator<Windowed<String>, Long> multiKey = windowStore.fetchAll(
    Instant.now().minus(Duration.ofHours(1)),
    Instant.now()
);

while (multiKey.hasNext()) {
    KeyValue<Windowed<String>, Long> kv = multiKey.next();
    System.out.println("Key: " + kv.key.key() + 
        ", Window: " + kv.key.window().start() + 
        ", Count: " + kv.value);
}
multiKey.close();

5. Distributed Queries

The Challenge

Problem:
├─ State partitioned across multiple instances
├─ Instance A has keys [a-m]
├─ Instance B has keys [n-z]
└─ Query for "user_xyz" may be on different instance

Solution:
├─ Metadata API: Find which instance has key
├─ HTTP/gRPC: Query remote instance
└─ Aggregate results

Metadata API

// Find which instance has a key
StreamsMetadata metadata = streams.metadataForKey(
    "user-counts",        // Store name
    "user123",            // Key
    Serdes.String().serializer()  // Key serializer
);

if (metadata != null) {
    System.out.println("Key is on host: " + metadata.host());
    System.out.println("Port: " + metadata.port());
    System.out.println("State store: " + metadata.stateStoreNames());
}

REST API for Remote Queries

// 1. Expose REST endpoint in each instance
@GetMapping("/counts/{key}")
public ResponseEntity<Long> getCount(@PathVariable String key) {
    // Check if key is local
    StreamsMetadata metadata = streams.metadataForKey(
        "user-counts",
        key,
        Serdes.String().serializer()
    );

    // If local, query locally
    if (metadata.host().equals(localHost)) {
        ReadOnlyKeyValueStore<String, Long> store = streams.store(...);
        Long count = store.get(key);
        return ResponseEntity.ok(count);
    } else {
        // Forward to remote instance
        String remoteUrl = "http://" + metadata.host() + ":" + metadata.port() + "/counts/" + key;
        RestTemplate restTemplate = new RestTemplate();
        Long remoteCount = restTemplate.getForObject(remoteUrl, Long.class);
        return ResponseEntity.ok(remoteCount);
    }
}

Query All Instances

// Get metadata for all instances
Collection<StreamsMetadata> allMetadata = streams.allMetadataForStore("user-counts");

List<Long> allCounts = new ArrayList<>();

for (StreamsMetadata metadata : allMetadata) {
    String url = "http://" + metadata.host() + ":" + metadata.port() + "/counts/all";
    RestTemplate restTemplate = new RestTemplate();
    List<Long> counts = restTemplate.getForObject(url, List.class);
    allCounts.addAll(counts);
}

// Aggregate results
long total = allCounts.stream().mapToLong(Long::longValue).sum();

6. Query Limitations

What You CAN Do

✅ Query by key
✅ Range queries
✅ Windowed queries (time-based)
✅ Iterate all keys
✅ Count approximate entries

What You CANNOT Do

❌ Complex queries (JOINs, GROUP BY, etc.)
❌ Secondary indexes
❌ Full-text search
❌ Aggregations beyond what's materialized
❌ Transactions (read-only)

For advanced queries: Use ksqlDB or external database

7. Testing with TopologyTestDriver

What is TopologyTestDriver?

TopologyTestDriver = lightweight, in-memory test harness

Purpose:
├─ Unit test Kafka Streams topologies
├─ No real Kafka cluster needed
├─ Fast, deterministic tests
└─ Test processing logic in isolation

NOT for:
├─ Integration testing (use real Kafka)
├─ Performance testing
└─ Multi-instance behavior

Basic Setup

@Test
public void testTopology() {
    // Create topology
    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, String> input = builder.stream("input-topic");
    input.mapValues(v -> v.toUpperCase()).to("output-topic");
    Topology topology = builder.build();

    // Test properties
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");

    // Create test driver
    TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);

    // Test (see below)

    // Clean up
    testDriver.close();
}

Sending Test Input

// Create input topic
TestInputTopic<String, String> inputTopic = testDriver.createInputTopic(
    "input-topic",
    Serdes.String().serializer(),
    Serdes.String().serializer()
);

// Send records
inputTopic.pipeInput("key1", "hello");
inputTopic.pipeInput("key2", "world");

// With timestamp
inputTopic.pipeInput("key3", "test", Instant.now().toEpochMilli());

Reading Test Output

// Create output topic
TestOutputTopic<String, String> outputTopic = testDriver.createOutputTopic(
    "output-topic",
    Serdes.String().deserializer(),
    Serdes.String().deserializer()
);

// Read single record
KeyValue<String, String> result = outputTopic.readKeyValue();
assertEquals("key1", result.key);
assertEquals("HELLO", result.value);

// Read all records
List<KeyValue<String, String>> allResults = outputTopic.readKeyValuesToList();
assertEquals(2, allResults.size());

// Check if empty
assertTrue(outputTopic.isEmpty());

Testing Stateful Operations

@Test
public void testCount() {
    // Topology: count by key
    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, String> input = builder.stream("input");
    input.groupByKey()
         .count(Materialized.as("counts"))
         .toStream()
         .to("output");

    TopologyTestDriver testDriver = new TopologyTestDriver(builder.build(), props);

    TestInputTopic<String, String> inputTopic = ...;
    TestOutputTopic<String, Long> outputTopic = ...;

    // Send records
    inputTopic.pipeInput("user1", "event1");
    inputTopic.pipeInput("user1", "event2");
    inputTopic.pipeInput("user2", "event1");

    // Verify output
    assertEquals(1L, outputTopic.readValue());  // user1 -> 1
    assertEquals(2L, outputTopic.readValue());  // user1 -> 2
    assertEquals(1L, outputTopic.readValue());  // user2 -> 1
}

Querying State Stores in Tests

@Test
public void testStateStore() {
    // Build topology
    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, String> input = builder.stream("input");
    input.groupByKey().count(Materialized.as("counts"));

    TopologyTestDriver testDriver = new TopologyTestDriver(builder.build(), props);

    // Send records
    TestInputTopic<String, String> inputTopic = ...;
    inputTopic.pipeInput("user1", "event");

    // Query state store
    KeyValueStore<String, Long> store = testDriver.getKeyValueStore("counts");
    Long count = store.get("user1");
    assertEquals(1L, count);
}

Testing Time-Based Operations

@Test
public void testWindowing() {
    // Topology: 5-minute tumbling window
    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, String> input = builder.stream("input");
    input.groupByKey()
         .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
         .count()
         .toStream()
         .to("output");

    TopologyTestDriver testDriver = new TopologyTestDriver(builder.build(), props);

    TestInputTopic<String, String> inputTopic = ...;

    // Send with specific timestamps
    Instant time1 = Instant.parse("2023-01-01T10:00:00Z");
    Instant time2 = Instant.parse("2023-01-01T10:02:00Z");  // Same window
    Instant time3 = Instant.parse("2023-01-01T10:06:00Z");  // Different window

    inputTopic.pipeInput("user1", "event1", time1.toEpochMilli());
    inputTopic.pipeInput("user1", "event2", time2.toEpochMilli());
    inputTopic.pipeInput("user1", "event3", time3.toEpochMilli());

    // Verify: 2 records in first window, 1 in second
    TestOutputTopic<Windowed<String>, Long> outputTopic = ...;
    assertEquals(1L, outputTopic.readValue());  // First event
    assertEquals(2L, outputTopic.readValue());  // Second event (same window)
    assertEquals(1L, outputTopic.readValue());  // Third event (new window)
}

8. Integration Testing

With Embedded Kafka

@Test
public void testIntegration() {
    // Use Testcontainers or embedded Kafka
    KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.4.0")
    );
    kafka.start();

    // Configure Kafka Streams
    Properties props = new Properties();
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "integration-test");

    // Build and start streams
    KafkaStreams streams = new KafkaStreams(topology, props);
    streams.start();

    // Produce input
    Producer<String, String> producer = new KafkaProducer<>(...);
    producer.send(new ProducerRecord<>("input-topic", "key", "value"));

    // Consume output
    Consumer<String, String> consumer = new KafkaConsumer<>(...);
    consumer.subscribe(Collections.singletonList("output-topic"));
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));

    // Verify
    assertEquals(1, records.count());

    // Clean up
    streams.close();
    kafka.stop();
}

9. Best Practices

1. Name Stores Explicitly

// Good: Named stores
.count(Materialized.as("user-counts"));

// Bad: Auto-generated names (not queryable reliably)
.count();

2. Wait for RUNNING State

// Wait before querying
while (streams.state() != State.RUNNING) {
    Thread.sleep(100);
}

// Or use state listener
streams.setStateListener((newState, oldState) -> {
    if (newState == State.RUNNING) {
        // Now safe to query
    }
});

3. Handle Null Results

// Always check for null
Long count = store.get("user123");
if (count == null) {
    count = 0L;  // Default value
}

4. Close Iterators

// Always close iterators
KeyValueIterator<String, Long> iterator = store.all();
try {
    while (iterator.hasNext()) {
        // Process
    }
} finally {
    iterator.close();  // Prevent resource leak
}

5. Use TopologyTestDriver for Unit Tests

// Fast, deterministic tests
// No Kafka cluster needed
TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);
// Test logic
testDriver.close();

6. Test Edge Cases

// Test empty input
inputTopic.pipeInput(null, null);

// Test late-arriving data
inputTopic.pipeInput("key", "value", pastTimestamp);

// Test out-of-order data
inputTopic.pipeInput("key", "value2", time1);
inputTopic.pipeInput("key", "value1", time2);  // time2 > time1

10. Interactive Query Patterns

Pattern 1: Health Check Endpoint

@GetMapping("/health")
public ResponseEntity<String> health() {
    if (streams.state() == State.RUNNING) {
        return ResponseEntity.ok("Healthy");
    } else {
        return ResponseEntity.status(503).body("Not ready");
    }
}

Pattern 2: Aggregated Query

@GetMapping("/total-count")
public ResponseEntity<Long> totalCount() {
    ReadOnlyKeyValueStore<String, Long> store = streams.store(...);
    KeyValueIterator<String, Long> all = store.all();

    long total = 0;
    while (all.hasNext()) {
        total += all.next().value;
    }
    all.close();

    return ResponseEntity.ok(total);
}

Pattern 3: Top-N Query

@GetMapping("/top-users")
public ResponseEntity<List<UserCount>> topUsers(@RequestParam int n) {
    ReadOnlyKeyValueStore<String, Long> store = streams.store(...);
    KeyValueIterator<String, Long> all = store.all();

    PriorityQueue<UserCount> topN = new PriorityQueue<>(n, 
        Comparator.comparing(UserCount::count).reversed());

    while (all.hasNext()) {
        KeyValue<String, Long> kv = all.next();
        topN.add(new UserCount(kv.key, kv.value));
        if (topN.size() > n) {
            topN.poll();
        }
    }
    all.close();

    return ResponseEntity.ok(new ArrayList<>(topN));
}

Pattern 4: Caching Query Results

// Cache query results to reduce load
Cache<String, Long> cache = Caffeine.newBuilder()
    .expireAfterWrite(Duration.ofSeconds(10))
    .build();

@GetMapping("/counts/{key}")
public ResponseEntity<Long> getCount(@PathVariable String key) {
    return ResponseEntity.ok(cache.get(key, k -> {
        ReadOnlyKeyValueStore<String, Long> store = streams.store(...);
        return store.get(k);
    }));
}

11. Mental Model Summary

Interactive Queries

┌──────────────────────────────────────────────────┐
│  Kafka Streams Application                       │
│                                                  │
│  ┌────────────────┐      ┌────────────────┐     │
│  │  Processing    │─────→│  State Store   │     │
│  │  Topology      │      │  (Materialized)│     │
│  └────────────────┘      └────────┬───────┘     │
│                                   │             │
│                          ┌────────▼───────┐     │
│                          │ Interactive    │     │
│                          │ Query API      │     │
│                          └────────┬───────┘     │
└──────────────────────────────────┬──────────────┘
                                   │
                          ┌────────▼───────┐
                          │ REST/gRPC      │
                          │ Endpoint       │
                          └────────────────┘

State is queryable in real-time

Distributed Queries

┌─────────────────────────────────────────────┐
│  Client                                     │
│  └─ Query for "user123"                     │
└───────────────┬─────────────────────────────┘
                ↓
┌───────────────▼─────────────────────────────┐
│  Instance A (has keys a-m)                  │
│  ├─ Check metadata: "user123" on Instance B │
│  └─ Forward query to Instance B             │
└───────────────┬─────────────────────────────┘
                ↓
┌───────────────▼─────────────────────────────┐
│  Instance B (has keys n-z)                  │
│  ├─ Query local state store                 │
│  └─ Return result                           │
└─────────────────────────────────────────────┘

Metadata API finds correct instance

12. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ What are interactive queries?
✅ How do I enable a state store for querying?
✅ How do I query a key-value store?
✅ How do I query windowed state?
✅ How do I query state across multiple instances?
✅ What is TopologyTestDriver?
✅ How do I unit test stateful operations?
✅ What are the limitations of interactive queries?

13. What's Next?

Now that you understand interactive queries and testing, the next module will cover:

• Module 12.1: Monitoring & operating — metrics, dashboards, operational best practices
• Module 13.1: Advanced configuration — performance tuning and tool selection

Summary

You learned:

• ✅ Interactive queries expose state stores as queryable databases
• ✅ Use Materialized.as("store-name") to enable querying
• ✅ Query key-value stores with get(), range(), all()
• ✅ Query windowed stores with fetch() and time ranges
• ✅ Distributed queries use metadata API to find key location
• ✅ TopologyTestDriver for fast, in-memory unit tests (no Kafka needed)
• ✅ Test stateful operations by querying stores in tests
• ✅ Always close iterators to prevent resource leaks
• ✅ Wait for RUNNING state before querying
• ✅ IQ limitations: No complex queries, JOINs, or secondary indexes

Next up: Module 12.1 — Monitoring & Operating — Learn how to monitor, debug, and operate Kafka Streams applications in production.