Skip to content

Commit 0bc770d

Browse files
mjsaxmjsax
authored
MINOR: avoid WARN logs in KafkaStreams test (#18517)
Avoiding 'WARN Method #getTimestampedKeyValueStore() should be used to access a TimestampedKeyValueStore.' running KTableKTableForeignKeyJoinIntegrationTest. Reviewers: Bill Bejeck <[email protected]>
1 parent 92fd99b commit 0bc770d

File tree

1 file changed

+20
-17
lines changed

1 file changed

+20
-17
lines changed

‎streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyJoinIntegrationTest.java

+20-17
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import org.apache.kafka.streams.kstream.KTable;
3232
import org.apache.kafka.streams.kstream.Materialized;
3333
import org.apache.kafka.streams.kstream.ValueJoiner;
34+
import org.apache.kafka.streams.state.KeyValueIterator;
3435
import org.apache.kafka.streams.state.KeyValueStore;
3536
import org.apache.kafka.streams.state.Stores;
3637
import org.apache.kafka.streams.state.ValueAndTimestamp;
@@ -174,7 +175,7 @@ public void doJoinFromLeftThenDeleteLeftEntity(final boolean leftJoin,
174175
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
175176
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
176177
final TestOutputTopic<String, String> rejoinOutputTopic = rejoin ? driver.createOutputTopic(REJOIN_OUTPUT, new StringDeserializer(), new StringDeserializer()) : null;
177-
final KeyValueStore<String, String> store = driver.getKeyValueStore("store");
178+
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
178179

179180
// Pre-populate the RHS records. This test is all about what happens when we add/remove LHS records
180181
right.pipeInput("rhs1", "rhsValue1", baseTimestamp);
@@ -257,7 +258,7 @@ public void doJoinFromLeftThenDeleteLeftEntity(final boolean leftJoin,
257258
}
258259
// Now delete one LHS entity such that one delete is propagated down to the output.
259260

260-
left.pipeInput("lhs1", (String) null, baseTimestamp + 6);
261+
left.pipeInput("lhs1", null, baseTimestamp + 6);
261262
assertThat(
262263
outputTopic.readKeyValuesToMap(),
263264
is(mkMap(
@@ -298,7 +299,7 @@ public void doJoinFromRightThenDeleteRightEntity(final boolean leftJoin,
298299
final TestInputTopic<String, String> right = driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new StringSerializer());
299300
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
300301
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
301-
final KeyValueStore<String, String> store = driver.getKeyValueStore("store");
302+
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
302303

303304
// Pre-populate the LHS records. This test is all about what happens when we add/remove RHS records
304305
left.pipeInput("lhs1", "lhsValue1|rhs1", baseTimestamp);
@@ -381,7 +382,7 @@ public void doJoinFromRightThenDeleteRightEntity(final boolean leftJoin,
381382
}
382383

383384
// Now delete the RHS entity such that all matching keys have deletes propagated.
384-
right.pipeInput("rhs1", (String) null, baseTimestamp + 6);
385+
right.pipeInput("rhs1", null, baseTimestamp + 6);
385386

386387
assertThat(
387388
outputTopic.readKeyValuesToMap(),
@@ -417,7 +418,7 @@ public void shouldEmitTombstoneWhenDeletingNonJoiningRecords(final boolean leftJ
417418
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
418419
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
419420
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
420-
final KeyValueStore<String, String> store = driver.getKeyValueStore("store");
421+
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
421422

422423
left.pipeInput("lhs1", "lhsValue1|rhs1", baseTimestamp);
423424

@@ -439,7 +440,7 @@ public void shouldEmitTombstoneWhenDeletingNonJoiningRecords(final boolean leftJ
439440
// Deleting a non-joining record produces an unnecessary tombstone for inner joins, because
440441
// it's not possible to know whether a result was previously emitted.
441442
// For the left join, the tombstone is necessary.
442-
left.pipeInput("lhs1", (String) null, baseTimestamp + 1);
443+
left.pipeInput("lhs1", null, baseTimestamp + 1);
443444
{
444445
assertThat(
445446
outputTopic.readKeyValuesToMap(),
@@ -454,7 +455,7 @@ public void shouldEmitTombstoneWhenDeletingNonJoiningRecords(final boolean leftJ
454455
}
455456

456457
// Deleting a non-existing record is idempotent
457-
left.pipeInput("lhs1", (String) null, baseTimestamp + 2);
458+
left.pipeInput("lhs1", null, baseTimestamp + 2);
458459
{
459460
assertThat(
460461
outputTopic.readKeyValuesToMap(),
@@ -483,10 +484,10 @@ public void shouldNotEmitTombstonesWhenDeletingNonExistingRecords(final boolean
483484
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
484485
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
485486
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
486-
final KeyValueStore<String, String> store = driver.getKeyValueStore("store");
487+
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
487488

488489
// Deleting a record that never existed doesn't need to emit tombstones.
489-
left.pipeInput("lhs1", (String) null, baseTimestamp);
490+
left.pipeInput("lhs1", null, baseTimestamp);
490491
{
491492
assertThat(
492493
outputTopic.readKeyValuesToMap(),
@@ -516,7 +517,7 @@ public void joinShouldProduceNullsWhenValueHasNonMatchingForeignKey(final boolea
516517
final TestInputTopic<String, String> right = driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new StringSerializer());
517518
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
518519
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
519-
final KeyValueStore<String, String> store = driver.getKeyValueStore("store");
520+
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
520521

521522
left.pipeInput("lhs1", "lhsValue1|rhs1", baseTimestamp);
522523
// no output for a new inner join on a non-existent FK
@@ -623,7 +624,7 @@ public void shouldUnsubscribeOldForeignKeyIfLeftSideIsUpdated(final boolean left
623624
final TestInputTopic<String, String> right = driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new StringSerializer());
624625
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
625626
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
626-
final KeyValueStore<String, String> store = driver.getKeyValueStore("store");
627+
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
627628

628629
// Pre-populate the RHS records. This test is all about what happens when we change LHS records foreign key reference
629630
// then populate update on RHS
@@ -707,7 +708,7 @@ public void shouldEmitRecordOnNullForeignKeyForLeftJoins(final String optimizati
707708
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
708709
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
709710
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
710-
final KeyValueStore<String, String> store = driver.getKeyValueStore("store");
711+
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
711712

712713
left.pipeInput("lhs1", "lhsValue1|rhs1", baseTimestamp);
713714
{
@@ -744,11 +745,11 @@ public void shouldEmitRecordWhenOldAndNewFkDiffer(final String optimization,
744745
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
745746
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
746747
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
747-
final KeyValueStore<String, String> store = driver.getKeyValueStore("store");
748+
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
748749
final String subscriptionStoreName = driver.getAllStateStores().entrySet().stream()
749750
.filter(e -> e.getKey().contains("SUBSCRIPTION-STATE-STORE"))
750751
.findAny().orElseThrow(() -> new RuntimeException("couldn't find store")).getKey();
751-
final KeyValueStore<Bytes, ValueAndTimestamp<String>> subscriptionStore = driver.getKeyValueStore(subscriptionStoreName);
752+
final KeyValueStore<Bytes, ValueAndTimestamp<String>> subscriptionStore = driver.getTimestampedKeyValueStore(subscriptionStoreName);
752753
final Bytes key = subscriptionStoreKey("lhs1", "rhs1");
753754
left.pipeInput("lhs1", "lhsValue1|rhs1", baseTimestamp);
754755
{
@@ -786,9 +787,11 @@ private static Bytes subscriptionStoreKey(final String lhs, final String rhs) {
786787
return key;
787788
}
788789

789-
protected static Map<String, String> asMap(final KeyValueStore<String, String> store) {
790+
protected static Map<String, String> asMap(final KeyValueStore<String, ValueAndTimestamp<String>> store) {
790791
final HashMap<String, String> result = new HashMap<>();
791-
store.all().forEachRemaining(kv -> result.put(kv.key, kv.value));
792+
try (final KeyValueIterator<String, ValueAndTimestamp<String>> it = store.all()) {
793+
it.forEachRemaining(kv -> result.put(kv.key, kv.value.value()));
794+
}
792795
return result;
793796
}
794797

@@ -921,7 +924,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned(final boolean leftJoin,
921924
final TestInputTopic<String, String> right = driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new StringSerializer());
922925
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
923926
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
924-
final KeyValueStore<String, String> store = driver.getKeyValueStore("store");
927+
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
925928

926929
// RHS record
927930
right.pipeInput("rhs1", "rhsValue1", baseTimestamp + 4);

0 commit comments

Comments
 (0)