mjsax commented on a change in pull request #8706:
URL: https://github.com/apache/kafka/pull/8706#discussion_r431361485
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java
##########
@@ -296,6 +289,75 @@ public void shouldQuerySpecificStalePartitionStores()
throws Exception {
assertThat(store4.get(key), is(nullValue()));
}
+ @Test
+ public void shouldQuerySpecificStalePartitionStoresMultiStreamThreads()
throws Exception {
+ final int batch1NumMessages = 100;
+ final int key = 1;
+ final Semaphore semaphore = new Semaphore(0);
+ final int numStreamThreads = 2;
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(INPUT_TOPIC_NAME, Consumed.with(Serdes.Integer(),
Serdes.Integer()),
+ Materialized.<Integer, Integer, KeyValueStore<Bytes,
byte[]>>as(TABLE_NAME)
+ .withCachingDisabled())
+ .toStream()
+ .peek((k, v) -> semaphore.release());
+
+ final Properties streamsConfiguration1 = streamsConfiguration();
+ streamsConfiguration1.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,
numStreamThreads);
+
+ final Properties streamsConfiguration2 = streamsConfiguration();
+ streamsConfiguration2.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,
numStreamThreads);
+
+ final KafkaStreams kafkaStreams1 = createKafkaStreams(builder,
streamsConfiguration1);
+ final KafkaStreams kafkaStreams2 = createKafkaStreams(builder,
streamsConfiguration2);
+ final List<KafkaStreams> kafkaStreamsList =
Arrays.asList(kafkaStreams1, kafkaStreams2);
+
+ startApplicationAndWaitUntilRunning(kafkaStreamsList,
Duration.ofSeconds(60));
+
+ assertTrue(numStreamThreads > 1);
Review comment:
`numStreamThreads` is a `final` variable -> assertion can be removed
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/QueryableStoreProvider.java
##########
@@ -58,9 +58,21 @@ public QueryableStoreProvider(final
List<StreamThreadStateStoreProvider> storePr
}
final List<T> allStores = new ArrayList<>();
for (final StreamThreadStateStoreProvider storeProvider :
storeProviders) {
- allStores.addAll(storeProvider.stores(storeQueryParameters));
+ final List<T> stores = storeProvider.stores(storeQueryParameters);
+ if (stores != null && !stores.isEmpty()) {
+ allStores.addAll(stores);
+ if (storeQueryParameters.partition() != null) {
+ break;
+ }
+ }
}
if (allStores.isEmpty()) {
+ if (storeQueryParameters.partition() != null) {
Review comment:
We should extend `QueryableStoreProviderTest` for this case -- throwing
this exception moves from `StreamThreadStateStoreProvider` to here and we
should not just remove the test from `StreamThreadStateStoreProviderTest` but
add a new one to `QueryableStoreProviderTest`, too.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
