Skip to main content

Kafka Transport, KafkaRecordFetcher

KafkaRecordFetcher<K, V> is the only class in ktestify-core that knows about Kafka internals. It implements RecordFetcher<V> and is the sole implementation of the transport contract today.

Fetch lifecycle​

KafkaRecordFetcher.fetch()
│
├─ ① subscribeAndAwaitAssignment()
│    Subscribe to topic, poll until partitions are assigned (max attempts before timeout).
│
├─ ② calculateDeltaTime()
│    Priority (ms):
│      1. context.consumerDeltaTime  (from DataTable, already converted to ms)
│      2. properties["consumerDeltaTime"]  (seconds, converted here)
│      3. ktestify.framework.timeouts.consumer-delta-time  (HOCON duration)
│
├─ ③ seekToOffset(T₀ = now - deltaTime)
│    For each assigned partition:
│      offsetsForTimes(partition → T₀)  → seek to that offset
│      If no offset found after T₀ → seekToEnd
│
└─ ④ pollUntilRecordFound()
     SINGLE mode (isBatchConsumer = false):
       Return as soon as ≥1 record passes key filter + dedup check.
     BATCH mode (isBatchConsumer = true):
       Accumulate records until batchSize reached or timeout expires.
     Dedup check: MATCHED_RECORDS.contains(record.toMatchedRecord())
     Key filter:  context.expectedRecordKey matches record.key  (if set)

Deduplication registry​

// Static, shared across all KafkaRecordFetcher instances in the JVM
private static final Set<MatchedRecord> MATCHED_RECORDS =
    ConcurrentHashMap.newKeySet();

public static void clearMatchedRecords() {
    MATCHED_RECORDS.clear();
}

The registry prevents two steps in the same scenario from claiming the same Kafka record. It is cleared in the @Before hook via KtestifyHooks at the start of every Cucumber scenario.

warning

If you use KafkaRecordFetcher directly (outside Cucumber), you must call KafkaRecordFetcher.clearMatchedRecords() before each test, otherwise records from previous tests bleed through.

Creating a KafkaConsumer / KafkaProducer​

KafkaClientFactory creates Kafka client instances from ConsumerContext or ProducerContext properties. It is the only place in ktestify-core that calls new KafkaConsumer<>() / new KafkaProducer<>().

KafkaConsumer<String, String> consumer =
    KafkaClientFactory.createConsumer(context.getProperties());

KafkaProducer<String, String> producer =
    KafkaClientFactory.createProducer(context.getProperties());

Properties are derived from KtestifyConfig and merged with any per-context overrides from the DataTable.

Avro deserialisation​

AvroKafkaConsumer passes a KafkaAvroDeserializer (Confluent) as the value deserialiser. The Schema Registry URL and auth settings flow from SchemaRegistryConfig into the deserialiser properties automatically via KafkaClientFactory.

Error handling​

SituationException thrown
Timeout with 0 recordsFetchException("Timed out after Nms waiting for a record on topic '...'")
Timeout with partial batchFetchException("Timed out after Nms waiting for N record(s), only M collected.")
WakeupException during pollFetchException (logged, consumer closed cleanly)
Broker unreachableFetchException wrapping the Kafka exception

All FetchExceptions are caught by AbstractKafkaConsumer and re-thrown as ConsumerException.

See also​