You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2026/10/23 02:39:19 UTC
(kafka) branch trunk updated: KAFKA-21225: Add mirror exceptions (#23828)
This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new da222409669 KAFKA-21225: Add mirror exceptions (#23828)
da222409669 is described below
commit da2224096692906b5e4398b2531fc4e810550c01
Author: Federico Valeri <fe...@gmail.com>
AuthorDate: Fri Oct 23 04:39:11 2026 +0200
KAFKA-21225: Add mirror exceptions (#23828)
This patch adds new exceptions and protocol errors required by the
Cluster Mirroring feature. No exception is marked as public for now. We
can change that later if needed.
Reviewers: Ken Huang <s7...@gmail.com>, Luke Chen <sh...@gmail.com>
---------
Signed-off-by: Federico Valeri <fe...@gmail.com>
---
build.gradle | 11 ++++++++
.../ClusterMirrorAlreadyExistsException.java | 29 +++++++++++++++++++++
.../ClusterMirrorAuthorizationException.java | 29 +++++++++++++++++++++
.../common/errors/InvalidMirrorStateException.java | 29 +++++++++++++++++++++
.../errors/MirrorConfigNotAvailableException.java | 30 ++++++++++++++++++++++
.../common/errors/ReadOnlyTopicException.java | 30 ++++++++++++++++++++++
.../SourceMetadataNotAvailableException.java | 30 ++++++++++++++++++++++
.../TopicAlreadyInClusterMirrorException.java | 29 +++++++++++++++++++++
.../errors/TopicNotInClusterMirrorException.java | 29 +++++++++++++++++++++
.../errors/UnknownClusterMirrorException.java | 29 +++++++++++++++++++++
.../org/apache/kafka/common/protocol/Errors.java | 18 ++++++++++++-
11 files changed, 292 insertions(+), 1 deletion(-)
diff --git a/build.gradle b/build.gradle
index 06fdf0e5c45..5660bfff08a 100644
--- a/build.gradle
+++ b/build.gradle
@@ -2229,6 +2229,17 @@ project(':clients') {
include "**/org/apache/kafka/server/policy/*"
include "**/org/apache/kafka/server/quota/*"
include "**/org/apache/kafka/server/telemetry/*"
+ exclude(
+ 'org/apache/kafka/common/errors/ReadOnlyTopicException.java',
+ 'org/apache/kafka/common/errors/ClusterMirrorAuthorizationException.java',
+ 'org/apache/kafka/common/errors/UnknownClusterMirrorException.java',
+ 'org/apache/kafka/common/errors/TopicAlreadyInClusterMirrorException.java',
+ 'org/apache/kafka/common/errors/MirrorConfigNotAvailableException.java',
+ 'org/apache/kafka/common/errors/ClusterMirrorAlreadyExistsException.java',
+ 'org/apache/kafka/common/errors/SourceMetadataNotAvailableException.java',
+ 'org/apache/kafka/common/errors/TopicNotInClusterMirrorException.java',
+ 'org/apache/kafka/common/errors/InvalidMirrorStateException.java'
+ )
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/ClusterMirrorAlreadyExistsException.java b/clients/src/main/java/org/apache/kafka/common/errors/ClusterMirrorAlreadyExistsException.java
new file mode 100644
index 00000000000..0b33c823a2a
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/ClusterMirrorAlreadyExistsException.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+/**
+ * The cluster mirror with the given name already exists.
+ * This error occurs when attempting to create a cluster mirror that has already been created.
+ */
+public class ClusterMirrorAlreadyExistsException extends ApiException {
+ private static final long serialVersionUID = 1L;
+
+ public ClusterMirrorAlreadyExistsException(String message) {
+ super(message);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/ClusterMirrorAuthorizationException.java b/clients/src/main/java/org/apache/kafka/common/errors/ClusterMirrorAuthorizationException.java
new file mode 100644
index 00000000000..f7241feb5a2
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/ClusterMirrorAuthorizationException.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+/**
+ * The client is not authorized to perform the requested operation on the cluster mirror.
+ * This error occurs when cluster mirroring operations are attempted without the necessary permissions.
+ */
+public class ClusterMirrorAuthorizationException extends AuthorizationException {
+ private static final long serialVersionUID = 1L;
+
+ public ClusterMirrorAuthorizationException(String message) {
+ super(message);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidMirrorStateException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidMirrorStateException.java
new file mode 100644
index 00000000000..0a3b37e25ab
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/InvalidMirrorStateException.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+/**
+ * The cluster mirror is in an invalid state for the requested operation.
+ * This error occurs when attempting an operation on a cluster mirror that is not in the appropriate state.
+ */
+public class InvalidMirrorStateException extends ApiException {
+ private static final long serialVersionUID = 1L;
+
+ public InvalidMirrorStateException(String message) {
+ super(message);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/MirrorConfigNotAvailableException.java b/clients/src/main/java/org/apache/kafka/common/errors/MirrorConfigNotAvailableException.java
new file mode 100644
index 00000000000..8f34ec1619c
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/MirrorConfigNotAvailableException.java
@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+/**
+ * Mirror configuration metadata is not yet available on the broker.
+ * This error is retriable because the broker will eventually receive the metadata update.
+ * Clients should backoff and retry when this error is encountered.
+ */
+public class MirrorConfigNotAvailableException extends InvalidMetadataException {
+ private static final long serialVersionUID = 1L;
+
+ public MirrorConfigNotAvailableException(String message) {
+ super(message);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/ReadOnlyTopicException.java b/clients/src/main/java/org/apache/kafka/common/errors/ReadOnlyTopicException.java
new file mode 100644
index 00000000000..dbd7f46ca0f
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/ReadOnlyTopicException.java
@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+/**
+ * The topic cannot be written to because it is marked as read-only.
+ * This error typically occurs in cluster mirroring scenarios where a topic
+ * is designated as read-only to maintain consistency across mirrored clusters.
+ */
+public class ReadOnlyTopicException extends ApiException {
+ private static final long serialVersionUID = 1L;
+
+ public ReadOnlyTopicException(String message) {
+ super(message);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/SourceMetadataNotAvailableException.java b/clients/src/main/java/org/apache/kafka/common/errors/SourceMetadataNotAvailableException.java
new file mode 100644
index 00000000000..f6d436fe6b1
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/SourceMetadataNotAvailableException.java
@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+/**
+ * Source cluster metadata is not yet available during cluster mirroring.
+ * This error occurs when the source cluster metadata has not yet been retrieved.
+ * Clients should backoff and retry when this error is encountered.
+ */
+public class SourceMetadataNotAvailableException extends InvalidMetadataException {
+ private static final long serialVersionUID = 1L;
+
+ public SourceMetadataNotAvailableException(String message) {
+ super(message);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/TopicAlreadyInClusterMirrorException.java b/clients/src/main/java/org/apache/kafka/common/errors/TopicAlreadyInClusterMirrorException.java
new file mode 100644
index 00000000000..84c27360b67
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/TopicAlreadyInClusterMirrorException.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+/**
+ * The topic is already assigned to the specified cluster mirror.
+ * This error occurs when attempting to add a topic to a cluster mirror when it has already been assigned.
+ */
+public class TopicAlreadyInClusterMirrorException extends ApiException {
+ private static final long serialVersionUID = 1L;
+
+ public TopicAlreadyInClusterMirrorException(String message) {
+ super(message);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/TopicNotInClusterMirrorException.java b/clients/src/main/java/org/apache/kafka/common/errors/TopicNotInClusterMirrorException.java
new file mode 100644
index 00000000000..2402f195326
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/TopicNotInClusterMirrorException.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+/**
+ * The topic does not belong to the specified cluster mirror.
+ * This error occurs when attempting to access or modify a topic that is not assigned to the target cluster mirror.
+ */
+public class TopicNotInClusterMirrorException extends ApiException {
+ private static final long serialVersionUID = 1L;
+
+ public TopicNotInClusterMirrorException(String message) {
+ super(message);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/UnknownClusterMirrorException.java b/clients/src/main/java/org/apache/kafka/common/errors/UnknownClusterMirrorException.java
new file mode 100644
index 00000000000..9bc92a95ca0
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/UnknownClusterMirrorException.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+/**
+ * The cluster mirror with the specified name does not exist.
+ * This error occurs when attempting to access or modify a cluster mirror that has not been created.
+ */
+public class UnknownClusterMirrorException extends ApiException {
+ private static final long serialVersionUID = 1L;
+
+ public UnknownClusterMirrorException(String message) {
+ super(message);
+ }
+}
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 5b49ed383cd..a23b2d8c632 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
@@ -21,6 +21,8 @@ import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.errors.BrokerIdNotRegisteredException;
import org.apache.kafka.common.errors.BrokerNotAvailableException;
import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.errors.ClusterMirrorAlreadyExistsException;
+import org.apache.kafka.common.errors.ClusterMirrorAuthorizationException;
import org.apache.kafka.common.errors.ConcurrentTransactionsException;
import org.apache.kafka.common.errors.ControllerIdNotRegisteredException;
import org.apache.kafka.common.errors.ControllerMovedException;
@@ -63,6 +65,7 @@ import org.apache.kafka.common.errors.InvalidConfigurationException;
import org.apache.kafka.common.errors.InvalidFetchSessionEpochException;
import org.apache.kafka.common.errors.InvalidFetchSizeException;
import org.apache.kafka.common.errors.InvalidGroupIdException;
+import org.apache.kafka.common.errors.InvalidMirrorStateException;
import org.apache.kafka.common.errors.InvalidPartitionsException;
import org.apache.kafka.common.errors.InvalidPidMappingException;
import org.apache.kafka.common.errors.InvalidPrincipalTypeException;
@@ -87,6 +90,7 @@ import org.apache.kafka.common.errors.LeaderNotAvailableException;
import org.apache.kafka.common.errors.ListenerNotFoundException;
import org.apache.kafka.common.errors.LogDirNotFoundException;
import org.apache.kafka.common.errors.MemberIdRequiredException;
+import org.apache.kafka.common.errors.MirrorConfigNotAvailableException;
import org.apache.kafka.common.errors.MismatchedEndpointTypeException;
import org.apache.kafka.common.errors.NetworkException;
import org.apache.kafka.common.errors.NewLeaderElectedException;
@@ -107,6 +111,7 @@ import org.apache.kafka.common.errors.PositionOutOfRangeException;
import org.apache.kafka.common.errors.PreferredLeaderNotAvailableException;
import org.apache.kafka.common.errors.PrincipalDeserializationException;
import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.errors.ReadOnlyTopicException;
import org.apache.kafka.common.errors.ReassignmentInProgressException;
import org.apache.kafka.common.errors.RebalanceInProgressException;
import org.apache.kafka.common.errors.RebootstrapRequiredException;
@@ -129,14 +134,17 @@ import org.apache.kafka.common.errors.StreamsTopologyFencedException;
import org.apache.kafka.common.errors.TelemetryTooLargeException;
import org.apache.kafka.common.errors.ThrottlingQuotaExceededException;
import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.errors.TopicAlreadyInClusterMirrorException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.errors.TopicDeletionDisabledException;
import org.apache.kafka.common.errors.TopicExistsException;
+import org.apache.kafka.common.errors.TopicNotInClusterMirrorException;
import org.apache.kafka.common.errors.TransactionAbortableException;
import org.apache.kafka.common.errors.TransactionCoordinatorFencedException;
import org.apache.kafka.common.errors.TransactionalIdAuthorizationException;
import org.apache.kafka.common.errors.TransactionalIdNotFoundException;
import org.apache.kafka.common.errors.UnacceptableCredentialException;
+import org.apache.kafka.common.errors.UnknownClusterMirrorException;
import org.apache.kafka.common.errors.UnknownControllerIdException;
import org.apache.kafka.common.errors.UnknownLeaderEpochException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
@@ -424,7 +432,15 @@ public enum Errors {
SHARE_SESSION_LIMIT_REACHED(133, "The limit of share sessions has been reached.", ShareSessionLimitReachedException::new),
GROUP_DELETION_FAILED(134, "DeleteGroups could not complete; see the error message on the per-group result for details.", GroupDeletionFailedException::new),
STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED(135, "The broker could not process the topology description update; see the error message for details.", StreamsTopologyDescriptionUpdateFailedException::new),
- CONTROLLER_ID_NOT_REGISTERED(136, "The given controller ID was not registered.", ControllerIdNotRegisteredException::new);
+ CONTROLLER_ID_NOT_REGISTERED(136, "The given controller ID was not registered.", ControllerIdNotRegisteredException::new),
+ READ_ONLY_TOPIC(137, "The topic is read-only.", ReadOnlyTopicException::new),
+ CLUSTER_MIRROR_ALREADY_EXISTS(138, "The cluster mirror already exists.", ClusterMirrorAlreadyExistsException::new),
+ UNKNOWN_CLUSTER_MIRROR(139, "The cluster mirror does not exist.", UnknownClusterMirrorException::new),
+ TOPIC_ALREADY_IN_CLUSTER_MIRROR(140, "The topic is already assigned to a cluster mirror.", TopicAlreadyInClusterMirrorException::new),
+ TOPIC_NOT_IN_CLUSTER_MIRROR(141, "The topic does not belong to the specified cluster mirror.", TopicNotInClusterMirrorException::new),
+ CLUSTER_MIRROR_AUTHORIZATION_FAILED(142, "Cluster mirror authorization failed.", ClusterMirrorAuthorizationException::new),
+ INVALID_CLUSTER_MIRROR_STATE(143, "Invalid cluster mirror state.", InvalidMirrorStateException::new),
+ MIRROR_CONFIG_NOT_AVAILABLE(144, "Mirror config is not available.", MirrorConfigNotAvailableException::new);
private static final Logger log = LoggerFactory.getLogger(Errors.class);