From 12ae09000a8ff0ed3e5a7dd4e47b41b5560544a1 Mon Sep 17 00:00:00 2001 From: vitojeng Date: Thu, 11 Jun 2020 09:11:51 +0800 Subject: [PATCH 1/3] KAFKA-5876: IQ should throw different exceptions for different errors(part 1) Add new sub-classes of InvalidStateStoreException --- .../errors/InvalidStateStoreException.java | 12 +++--- .../InvalidStateStorePartitionException.java | 35 ++++++++++++++++++ .../errors/StateStoreMigratedException.java | 37 +++++++++++++++++++ .../StateStoreNotAvailableException.java | 37 +++++++++++++++++++ .../errors/StreamsNotStartedException.java | 34 +++++++++++++++++ .../errors/StreamsRebalancingException.java | 36 ++++++++++++++++++ .../errors/UnknownStateStoreException.java | 34 +++++++++++++++++ 7 files changed, 218 insertions(+), 7 deletions(-) create mode 100644 streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStorePartitionException.java create mode 100644 streams/src/main/java/org/apache/kafka/streams/errors/StateStoreMigratedException.java create mode 100644 streams/src/main/java/org/apache/kafka/streams/errors/StateStoreNotAvailableException.java create mode 100644 streams/src/main/java/org/apache/kafka/streams/errors/StreamsNotStartedException.java create mode 100644 streams/src/main/java/org/apache/kafka/streams/errors/StreamsRebalancingException.java create mode 100644 streams/src/main/java/org/apache/kafka/streams/errors/UnknownStateStoreException.java 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..107d0ff2261c6 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,11 @@ /** - * 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}. + * InvalidStateStoreException not thrown directly but only following sub-classes:

+ * {@link StreamsNotStartedException}, {@link StreamsRebalancingException}, + * {@link StateStoreMigratedException}, {@link StateStoreNotAvailableException}, + * {@link UnknownStateStoreException}, {@link InvalidStateStorePartitionException} */ 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..b198ee324e94a --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStorePartitionException.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 specific state store being queried via + * {@link org.apache.kafka.streams.StoreQueryParameters} used an invalid 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..134fa652935bd --- /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}. + * It could happen because the partition 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..5de74aa7e8ca1 --- /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. It 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..2c81231238424 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsNotStartedException.java @@ -0,0 +1,34 @@ +/* + * 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; + +/** + * Indicate query a state store when Kafka Streams state is {@link org.apache.kafka.streams.KafkaStreams.State#CREATED CREATED}. + * User can just retry and wait until to {@link org.apache.kafka.streams.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..08e16b64a159c --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsRebalancingException.java @@ -0,0 +1,36 @@ +/* + * 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; + +/** + * Indicate query a state store and stream thread state is not running when Kafka Streams state is + * {@link org.apache.kafka.streams.KafkaStreams.State#RUNNING RUNNING} or + * {@link org.apache.kafka.streams.KafkaStreams.State#REBALANCING REBALANCING}. + * User can just retry and wait until rebalance finished. + */ +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..e72e82af164b4 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/errors/UnknownStateStoreException.java @@ -0,0 +1,34 @@ +/* + * 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 invalid. Hence, it will be futile to retry again. + */ +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); + } + +} From f3136d4c3bf74c09c039ae4df2cc070ebb743a09 Mon Sep 17 00:00:00 2001 From: vitojeng Date: Sat, 18 Jul 2020 23:24:46 +0800 Subject: [PATCH 2/3] address comments --- .../kafka/streams/errors/InvalidStateStoreException.java | 5 +---- .../errors/InvalidStateStorePartitionException.java | 5 ++++- .../kafka/streams/errors/StateStoreMigratedException.java | 2 +- .../streams/errors/StateStoreNotAvailableException.java | 2 +- .../kafka/streams/errors/StreamsNotStartedException.java | 7 +++++-- .../kafka/streams/errors/StreamsRebalancingException.java | 7 +++---- .../kafka/streams/errors/UnknownStateStoreException.java | 3 ++- 7 files changed, 17 insertions(+), 14 deletions(-) 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 107d0ff2261c6..3f64eedf71295 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 @@ -19,10 +19,7 @@ /** *

Indicates that there was a problem when trying to access a {@link org.apache.kafka.streams.processor.StateStore StateStore}. - * InvalidStateStoreException not thrown directly but only following sub-classes:

- * {@link StreamsNotStartedException}, {@link StreamsRebalancingException}, - * {@link StateStoreMigratedException}, {@link StateStoreNotAvailableException}, - * {@link UnknownStateStoreException}, {@link InvalidStateStorePartitionException} + * {@code InvalidStateStoreException} not thrown directly but only 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 index b198ee324e94a..60c68b9ea2c11 100644 --- a/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStorePartitionException.java +++ b/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStorePartitionException.java @@ -16,9 +16,12 @@ */ 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 an invalid partition. + * {@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 { 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 index 134fa652935bd..45329c8101b21 100644 --- a/streams/src/main/java/org/apache/kafka/streams/errors/StateStoreMigratedException.java +++ b/streams/src/main/java/org/apache/kafka/streams/errors/StateStoreMigratedException.java @@ -20,7 +20,7 @@ * 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}. - * It could happen because the partition moved to some other instance during a rebalance so + * 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 { 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 index 5de74aa7e8ca1..7cec17c40d6fa 100644 --- a/streams/src/main/java/org/apache/kafka/streams/errors/StateStoreNotAvailableException.java +++ b/streams/src/main/java/org/apache/kafka/streams/errors/StateStoreNotAvailableException.java @@ -17,7 +17,7 @@ package org.apache.kafka.streams.errors; /** - * Indicates that the state store being queried is already closed. It could happen when Kafka Streams is in + * 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. 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 index 2c81231238424..6820bd0ed72d0 100644 --- a/streams/src/main/java/org/apache/kafka/streams/errors/StreamsNotStartedException.java +++ b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsNotStartedException.java @@ -16,9 +16,12 @@ */ package org.apache.kafka.streams.errors; +import org.apache.kafka.streams.KafkaStreams; + /** - * Indicate query a state store when Kafka Streams state is {@link org.apache.kafka.streams.KafkaStreams.State#CREATED CREATED}. - * User can just retry and wait until to {@link org.apache.kafka.streams.KafkaStreams.State#RUNNING RUNNING} + * 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 to {@link KafkaStreams.State#RUNNING RUNNING} */ public class StreamsNotStartedException extends InvalidStateStoreException { 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 index 08e16b64a159c..4b8e14c9b6590 100644 --- a/streams/src/main/java/org/apache/kafka/streams/errors/StreamsRebalancingException.java +++ b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsRebalancingException.java @@ -17,10 +17,9 @@ package org.apache.kafka.streams.errors; /** - * Indicate query a state store and stream thread state is not running when Kafka Streams state is - * {@link org.apache.kafka.streams.KafkaStreams.State#RUNNING RUNNING} or - * {@link org.apache.kafka.streams.KafkaStreams.State#REBALANCING REBALANCING}. - * User can just retry and wait until rebalance finished. + * 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 { 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 index e72e82af164b4..0ee0658bec4ad 100644 --- a/streams/src/main/java/org/apache/kafka/streams/errors/UnknownStateStoreException.java +++ b/streams/src/main/java/org/apache/kafka/streams/errors/UnknownStateStoreException.java @@ -17,7 +17,8 @@ package org.apache.kafka.streams.errors; /** - * Indicates that the state store being queried is invalid. Hence, it will be futile to retry again. + * 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 { From 6cf8ee4017df723614875ff131eedb5aa5a4cb14 Mon Sep 17 00:00:00 2001 From: vitojeng Date: Tue, 21 Jul 2020 13:30:17 +0800 Subject: [PATCH 3/3] address comments --- .../kafka/streams/errors/InvalidStateStoreException.java | 4 ++-- .../kafka/streams/errors/StreamsNotStartedException.java | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) 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 3f64eedf71295..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,8 +18,8 @@ /** - *

Indicates that there was a problem when trying to access a {@link org.apache.kafka.streams.processor.StateStore StateStore}. - * {@code InvalidStateStoreException} not thrown directly but only following sub-classes. + * 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/StreamsNotStartedException.java b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsNotStartedException.java index 6820bd0ed72d0..524e5f6765433 100644 --- a/streams/src/main/java/org/apache/kafka/streams/errors/StreamsNotStartedException.java +++ b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsNotStartedException.java @@ -21,7 +21,7 @@ /** * 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 to {@link KafkaStreams.State#RUNNING RUNNING} + * You can retry to query the state after the state transitioned to {@link KafkaStreams.State#RUNNING RUNNING}. */ public class StreamsNotStartedException extends InvalidStateStoreException {