KAFKA-5876: IQ should throw different exceptions for different errors(part 1)

streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java

@@ -18,13 +18,8 @@
 
 
 /**
- * Indicates that there was a problem when trying to access a
- * {@link org.apache.kafka.streams.processor.StateStore StateStore}, i.e, the Store is no longer valid because it is
- * closed or doesn't exist any more due to a rebalance.
- * <p>
- * These exceptions may be transient, i.e., during a rebalance it won't be possible to query the stores as they are
- * being (re)-initialized. Once the rebalance has completed the stores will be available again. Hence, it is valid
- * to backoff and retry when handling this exception.
+ * Indicates that there was a problem when trying to access a {@link org.apache.kafka.streams.processor.StateStore StateStore}.
+ * {@code InvalidStateStoreException} is not thrown directly but only its following sub-classes.
  */
 public class InvalidStateStoreException extends StreamsException {
 

streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStorePartitionException.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors;
+
+import org.apache.kafka.streams.KafkaStreams;
+
+/**
+ * Indicates that the specific state store being queried via
+ * {@link org.apache.kafka.streams.StoreQueryParameters} used a partitioning that is not assigned to this instance.
+ * You can use {@link KafkaStreams#allMetadata()} to discover the correct instance that hosts the requested partition.
+ */
+public class InvalidStateStorePartitionException extends InvalidStateStoreException {
+
+    private static final long serialVersionUID = 1L;
+
+    public InvalidStateStorePartitionException(final String message) {
+        super(message);
+    }
+
+    public InvalidStateStorePartitionException(final String message, final Throwable throwable) {
+        super(message, throwable);
+    }
+
+}

streams/src/main/java/org/apache/kafka/streams/errors/StateStoreMigratedException.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors;
+
+/**
+ * Indicates that the state store being queried is closed although the Kafka Streams state is
+ * {@link org.apache.kafka.streams.KafkaStreams.State#RUNNING RUNNING} or
+ * {@link org.apache.kafka.streams.KafkaStreams.State#REBALANCING REBALANCING}.
+ * This could happen because the store moved to some other instance during a rebalance so
+ * rediscovery of the state store is required before retrying.
+ */
+public class StateStoreMigratedException extends InvalidStateStoreException {
+
+    private static final long serialVersionUID = 1L;
+
+    public StateStoreMigratedException(final String message) {
+        super(message);
+    }
+
+    public StateStoreMigratedException(final String message, final Throwable throwable) {
+        super(message, throwable);
+    }
+}

streams/src/main/java/org/apache/kafka/streams/errors/StateStoreNotAvailableException.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors;
+
+/**
+ * Indicates that the state store being queried is already closed. This could happen when Kafka Streams is in
+ * {@link org.apache.kafka.streams.KafkaStreams.State#PENDING_SHUTDOWN PENDING_SHUTDOWN} or
+ * {@link org.apache.kafka.streams.KafkaStreams.State#NOT_RUNNING NOT_RUNNING} or
+ * {@link org.apache.kafka.streams.KafkaStreams.State#ERROR ERROR} state.
+ */
+public class StateStoreNotAvailableException extends InvalidStateStoreException {
+
+    private static final long serialVersionUID = 1L;
+
+    public StateStoreNotAvailableException(final String message) {
+        super(message);
+    }
+
+    public StateStoreNotAvailableException(final String message, final Throwable throwable) {
+        super(message, throwable);
+    }
+
+}

streams/src/main/java/org/apache/kafka/streams/errors/StreamsNotStartedException.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors;
+
+import org.apache.kafka.streams.KafkaStreams;
+
+/**
+ * Indicates that Kafka Streams is in state {@link KafkaStreams.State#CREATED CREATED} and thus state stores cannot be queries yet.
+ * To query state stores, it's required to first start Kafka Streams via {@link KafkaStreams#start()}.
+ * You can retry to query the state after the state transitioned to {@link KafkaStreams.State#RUNNING RUNNING}.
+ */
+public class StreamsNotStartedException extends InvalidStateStoreException {
+
+    private static final long serialVersionUID = 1L;
+
+    public StreamsNotStartedException(final String message) {
+        super(message);
+    }
+
+    public StreamsNotStartedException(final String message, final Throwable throwable) {
+        super(message, throwable);
+    }
+}

streams/src/main/java/org/apache/kafka/streams/errors/StreamsRebalancingException.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors;
+
+/**
+ * Indicates that Kafka Streams is in state {@link org.apache.kafka.streams.KafkaStreams.State#REBALANCING REBALANCING} and thus
+ * cannot be queried by default. You can retry to query after the rebalance finished. As an alternative, you can also query
+ * (potentially stale) state stores during a rebalance via {@link org.apache.kafka.streams.StoreQueryParameters#enableStaleStores()}.
+ */
+public class StreamsRebalancingException extends InvalidStateStoreException {
+
+    private static final long serialVersionUID = 1L;
+
+    public StreamsRebalancingException(final String message) {
+        super(message);
+    }
+
+    public StreamsRebalancingException(final String message, final Throwable throwable) {
+        super(message, throwable);
+    }
+}

streams/src/main/java/org/apache/kafka/streams/errors/UnknownStateStoreException.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors;
+
+/**
+ * Indicates that the state store being queried is unknown, i.e., the state store does either not exist in your topology
+ * or it is not queryable.
+ */
+public class UnknownStateStoreException extends InvalidStateStoreException {
+
+    private static final long serialVersionUID = 1L;
+
+    public UnknownStateStoreException(final String message) {
+        super(message);
+    }
+
+    public UnknownStateStoreException(final String message, final Throwable throwable) {
+        super(message, throwable);
+    }
+
+}