Correctly handle null keys and null values when draining table output
@@ -339,12 +339,18 @@ public class IntegrationTestUtils {
final Map<K, V> results = new LinkedHashMap<>();
while (true) {
final ProducerRecord<K, V> record = topologyTestDriver.readOutput(topic, keyDeserializer, valueDeserializer);
- if (record == null) {
+ // Tables ignore records with null keys.
+ if (record == null || record.key() == null) {
break;
+ } else {
+ // For tables, a null-valued record represents a "DELETE" or tombstone for the corresponding key.
+ if (record.value() == null) {
+ results.remove(record.key());
} else {
results.put(record.key(), record.value());
}
}
+ }
return results;
}
