KAFKA-5876: KIP-216 Part 4, Apply InvalidStateStorePartitionException for Interactive Queries (#10657)

KIP-216, part 4 - apply InvalidStateStorePartitionException

Reviewers: Anna Sophie Blee-Goldman <[email protected]>

Author: a0x8o

docs/streams/upgrade-guide.html

@@ -101,6 +101,7 @@ <h3><a id="streams_api_changes_300" href="#streams_api_changes_300">Streams API
     <ul>
         <li> <code>UnknownStateStoreException</code>: If the specified store name does not exist in the topology, an <code>UnknownStateStoreException</code> will be thrown instead of the former <code>InvalidStateStoreException</code>.</li>
         <li> <code>StreamsNotStartedException</code>: If Streams state is <code>CREATED</code>, a <code>StreamsNotStartedException</code> will be thrown.</li>
+        <li> <code>InvalidStateStorePartitionException</code>: If the specified partition does not exist, a <code>InvalidStateStorePartitionException</code> will be thrown.</li>
     </ul>
     <p>
         See <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-216%3A+IQ+should+throw+different+exceptions+for+different+errors">KIP-216</a> for more information.

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -46,6 +46,7 @@
 import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
 import org.apache.kafka.streams.errors.TopologyException;
 import org.apache.kafka.streams.errors.UnknownStateStoreException;
+import org.apache.kafka.streams.errors.InvalidStateStorePartitionException;
 import org.apache.kafka.streams.internals.metrics.ClientMetrics;
 import org.apache.kafka.streams.processor.Processor;
 import org.apache.kafka.streams.processor.StateRestoreListener;
@@ -1526,6 +1527,7 @@ public <K> KeyQueryMetadata queryMetadataForKey(final String storeName,
      * @throws StreamsNotStartedException If Streams state is {@link KafkaStreams.State#CREATED CREATED}. Just
      *         retry and wait until to {@link KafkaStreams.State#RUNNING RUNNING}.
      * @throws UnknownStateStoreException If the specified store name does not exist in the topology.
+     * @throws InvalidStateStorePartitionException If the specified partition does not exist.
      * @throws InvalidStateStoreException If the Streams instance isn't in a queryable state.
      *                                    If the store's type does not match the QueryableStoreType,
      *                                    the Streams instance is not in a queryable state with respect

streams/src/main/java/org/apache/kafka/streams/state/internals/WrappingStoreProvider.java

@@ -18,6 +18,7 @@
 
 import org.apache.kafka.streams.StoreQueryParameters;
 import org.apache.kafka.streams.errors.InvalidStateStoreException;
+import org.apache.kafka.streams.errors.InvalidStateStorePartitionException;
 import org.apache.kafka.streams.state.QueryableStoreType;
 
 import java.util.ArrayList;
@@ -57,7 +58,7 @@ public <T> List<T> stores(final String storeName,
         }
         if (allStores.isEmpty()) {
             if (storeQueryParameters.partition() != null) {
-                throw new InvalidStateStoreException(
+                throw new InvalidStateStorePartitionException(
                         String.format("The specified partition %d for store %s does not exist.",
                                 storeQueryParameters.partition(),
                                 storeName));

streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java

@@ -74,6 +74,7 @@
 import static org.easymock.EasyMock.mock;
 import static org.easymock.EasyMock.notNull;
 import static org.easymock.EasyMock.reset;
+import static org.hamcrest.CoreMatchers.either;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.notNullValue;
@@ -439,6 +440,8 @@ public void shouldReturnUUIDsWithStringPrefix() {
         final UUID uuid1 = UUID.randomUUID();
         final UUID uuid2 = UUID.randomUUID();
         final String prefix = uuid1.toString().substring(0, 4);
+        final int numMatches = uuid2.toString().substring(0, 4).equals(prefix) ? 2 : 1;
+
         entries.add(new KeyValue<>(
             new Bytes(uuidSerializer.serialize(null, uuid1)),
             stringSerializer.serialize(null, "a")));
@@ -460,8 +463,12 @@ public void shouldReturnUUIDsWithStringPrefix() {
             numberOfKeysReturned++;
         }
 
-        assertThat(numberOfKeysReturned, is(1));
-        assertThat(valuesWithPrefix.get(0), is("a"));
+        assertThat(numberOfKeysReturned, is(numMatches));
+        if (numMatches == 2) {
+            assertThat(valuesWithPrefix.get(0), either(is("a")).or(is("b")));
+        } else {
+            assertThat(valuesWithPrefix.get(0), is("a"));
+        }
     }
 
     @Test

streams/src/test/java/org/apache/kafka/streams/state/internals/WrappingStoreProviderTest.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.StoreQueryParameters;
 import org.apache.kafka.streams.errors.InvalidStateStoreException;
+import org.apache.kafka.streams.errors.InvalidStateStorePartitionException;
 import org.apache.kafka.streams.state.NoOpWindowStore;
 import org.apache.kafka.streams.state.QueryableStoreTypes;
 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
@@ -82,6 +83,13 @@ public void shouldThrowInvalidStoreExceptionIfNoStoreOfTypeFound() {
         assertThrows(InvalidStateStoreException.class, () -> wrappingStoreProvider.stores("doesn't exist", QueryableStoreTypes.<String, String>keyValueStore()));
     }
 
+    @Test
+    public void shouldThrowInvalidStoreExceptionIfNoPartitionFound() {
+        final int invalidPartition = numStateStorePartitions + 1;
+        wrappingStoreProvider.setStoreQueryParameters(StoreQueryParameters.fromNameAndType("kv", QueryableStoreTypes.<String, String>keyValueStore()).withPartition(invalidPartition));
+        assertThrows(InvalidStateStorePartitionException.class, () -> wrappingStoreProvider.stores("kv", QueryableStoreTypes.<String, String>keyValueStore()));
+    }
+
     @Test
     public void shouldReturnAllStoreWhenQueryWithoutPartition() {
         wrappingStoreProvider.setStoreQueryParameters(StoreQueryParameters.fromNameAndType("kv", QueryableStoreTypes.<String, String>keyValueStore()));