Repository: kafka Updated Branches: refs/heads/trunk 401ae7707 -> cc3570d1a
KAFKA-2999: Errors enum should be a 1 to 1 mapping of error codes and⦠⦠exceptions Author: Grant Henke <[email protected]> Reviewers: Gwen Shapira Closes #766 from granthenke/errors-map Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/cc3570d1 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/cc3570d1 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/cc3570d1 Branch: refs/heads/trunk Commit: cc3570d1a28354acd082c41a15d9c9efde540b16 Parents: 401ae77 Author: Grant Henke <[email protected]> Authored: Mon Jan 18 10:09:17 2016 -0800 Committer: Gwen Shapira <[email protected]> Committed: Mon Jan 18 10:09:17 2016 -0800 ---------------------------------------------------------------------- .../InconsistentGroupProtocolException.java | 25 ++++++++ .../InvalidCommitOffsetSizeException.java | 25 ++++++++ .../common/errors/InvalidGroupIdException.java | 25 ++++++++ .../errors/InvalidSessionTimeoutException.java | 25 ++++++++ .../apache/kafka/common/protocol/Errors.java | 12 ++-- .../kafka/common/protocol/ErrorsTest.java | 63 ++++++++++++++++++++ 6 files changed, 171 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/cc3570d1/clients/src/main/java/org/apache/kafka/common/errors/InconsistentGroupProtocolException.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InconsistentGroupProtocolException.java b/clients/src/main/java/org/apache/kafka/common/errors/InconsistentGroupProtocolException.java new file mode 100644 index 0000000..f3ccce0 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/errors/InconsistentGroupProtocolException.java @@ -0,0 +1,25 @@ +/** + * 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.common.errors; + +public class InconsistentGroupProtocolException extends ApiException { + private static final long serialVersionUID = 1L; + + public InconsistentGroupProtocolException(String message, Throwable cause) { + super(message, cause); + } + + public InconsistentGroupProtocolException(String message) { + super(message); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/cc3570d1/clients/src/main/java/org/apache/kafka/common/errors/InvalidCommitOffsetSizeException.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidCommitOffsetSizeException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidCommitOffsetSizeException.java new file mode 100644 index 0000000..9573ad2 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/errors/InvalidCommitOffsetSizeException.java @@ -0,0 +1,25 @@ +/** + * 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.common.errors; + +public class InvalidCommitOffsetSizeException extends ApiException { + private static final long serialVersionUID = 1L; + + public InvalidCommitOffsetSizeException(String message, Throwable cause) { + super(message, cause); + } + + public InvalidCommitOffsetSizeException(String message) { + super(message); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/cc3570d1/clients/src/main/java/org/apache/kafka/common/errors/InvalidGroupIdException.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidGroupIdException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidGroupIdException.java new file mode 100644 index 0000000..191f456 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/errors/InvalidGroupIdException.java @@ -0,0 +1,25 @@ +/** + * 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.common.errors; + +public class InvalidGroupIdException extends ApiException { + private static final long serialVersionUID = 1L; + + public InvalidGroupIdException(String message, Throwable cause) { + super(message, cause); + } + + public InvalidGroupIdException(String message) { + super(message); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/cc3570d1/clients/src/main/java/org/apache/kafka/common/errors/InvalidSessionTimeoutException.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidSessionTimeoutException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidSessionTimeoutException.java new file mode 100644 index 0000000..3ee411e --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/errors/InvalidSessionTimeoutException.java @@ -0,0 +1,25 @@ +/** + * 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.common.errors; + +public class InvalidSessionTimeoutException extends ApiException { + private static final long serialVersionUID = 1L; + + public InvalidSessionTimeoutException(String message, Throwable cause) { + super(message, cause); + } + + public InvalidSessionTimeoutException(String message) { + super(message); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/cc3570d1/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java index c635017..8581544 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java @@ -28,8 +28,12 @@ import org.apache.kafka.common.errors.GroupAuthorizationException; import org.apache.kafka.common.errors.GroupCoordinatorNotAvailableException; import org.apache.kafka.common.errors.GroupLoadInProgressException; import org.apache.kafka.common.errors.IllegalGenerationException; +import org.apache.kafka.common.errors.InconsistentGroupProtocolException; +import org.apache.kafka.common.errors.InvalidCommitOffsetSizeException; import org.apache.kafka.common.errors.InvalidFetchSizeException; +import org.apache.kafka.common.errors.InvalidGroupIdException; import org.apache.kafka.common.errors.InvalidRequiredAcksException; +import org.apache.kafka.common.errors.InvalidSessionTimeoutException; import org.apache.kafka.common.errors.InvalidTopicException; import org.apache.kafka.common.errors.LeaderNotAvailableException; import org.apache.kafka.common.errors.NetworkException; @@ -105,17 +109,17 @@ public enum Errors { ILLEGAL_GENERATION(22, new IllegalGenerationException("Specified group generation id is not valid.")), INCONSISTENT_GROUP_PROTOCOL(23, - new ApiException("The group member's supported protocols are incompatible with those of existing members.")), + new InconsistentGroupProtocolException("The group member's supported protocols are incompatible with those of existing members.")), INVALID_GROUP_ID(24, - new ApiException("The configured groupId is invalid")), + new InvalidGroupIdException("The configured groupId is invalid")), UNKNOWN_MEMBER_ID(25, new UnknownMemberIdException("The coordinator is not aware of this member.")), INVALID_SESSION_TIMEOUT(26, - new ApiException("The session timeout is not within an acceptable range.")), + new InvalidSessionTimeoutException("The session timeout is not within an acceptable range.")), REBALANCE_IN_PROGRESS(27, new RebalanceInProgressException("The group is rebalancing, so a rejoin is needed.")), INVALID_COMMIT_OFFSET_SIZE(28, - new ApiException("The committing offset data size is not valid")), + new InvalidCommitOffsetSizeException("The committing offset data size is not valid")), TOPIC_AUTHORIZATION_FAILED(29, new TopicAuthorizationException("Topic authorization failed.")), GROUP_AUTHORIZATION_FAILED(30, http://git-wip-us.apache.org/repos/asf/kafka/blob/cc3570d1/clients/src/test/java/org/apache/kafka/common/protocol/ErrorsTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/common/protocol/ErrorsTest.java b/clients/src/test/java/org/apache/kafka/common/protocol/ErrorsTest.java new file mode 100644 index 0000000..b511b4b --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/protocol/ErrorsTest.java @@ -0,0 +1,63 @@ +/** + * 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.common.protocol; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNull; + +import java.util.HashSet; +import java.util.Set; + +import org.apache.kafka.common.errors.ApiException; +import org.junit.Test; + +public class ErrorsTest { + + @Test + public void testUniqueErrorCodes() { + Set<Short> codeSet = new HashSet<>(); + for (Errors error : Errors.values()) { + codeSet.add(error.code()); + } + assertEquals("Error codes must be unique", codeSet.size(), Errors.values().length); + } + + @Test + public void testUniqueExceptions() { + Set<Class> exceptionSet = new HashSet<>(); + for (Errors error : Errors.values()) { + if (error != Errors.NONE) + exceptionSet.add(error.exception().getClass()); + } + assertEquals("Exceptions must be unique", exceptionSet.size(), Errors.values().length - 1); // Ignore NONE + } + + @Test + public void testExceptionsAreNotGeneric() { + for (Errors error : Errors.values()) { + if (error != Errors.NONE) + assertNotEquals("Generic ApiException should not be used", error.exception().getClass(), ApiException.class); + } + } + + @Test + public void testNoneException() { + assertNull("The NONE error should not have an exception", Errors.NONE.exception()); + } + +}
