diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java b/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java index 6f770736c622b..50c961b7af4f0 100644 --- a/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java +++ b/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. - *

- * 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 { diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStorePartitionException.java b/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStorePartitionException.java new file mode 100644 index 0000000000000..60c68b9ea2c11 --- /dev/null +++ b/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); + } + +} diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/StateStoreMigratedException.java b/streams/src/main/java/org/apache/kafka/streams/errors/StateStoreMigratedException.java new file mode 100644 index 0000000000000..45329c8101b21 --- /dev/null +++ b/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); + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/StateStoreNotAvailableException.java b/streams/src/main/java/org/apache/kafka/streams/errors/StateStoreNotAvailableException.java new file mode 100644 index 0000000000000..7cec17c40d6fa --- /dev/null +++ b/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); + } + +} diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/StreamsNotStartedException.java b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsNotStartedException.java new file mode 100644 index 0000000000000..524e5f6765433 --- /dev/null +++ b/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); + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/StreamsRebalancingException.java b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsRebalancingException.java new file mode 100644 index 0000000000000..4b8e14c9b6590 --- /dev/null +++ b/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); + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/UnknownStateStoreException.java b/streams/src/main/java/org/apache/kafka/streams/errors/UnknownStateStoreException.java new file mode 100644 index 0000000000000..0ee0658bec4ad --- /dev/null +++ b/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); + } + +}