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 <s7133700@gmail.com>, Luke Chen <showuon@gmail.com>

---------

Signed-off-by: Federico Valeri <fedevaleri@gmail.com>

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'
+    )
   }
 }
 

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);
+    }
+}

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);
+    }
+}

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);
+    }
+}

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);
+    }
+}

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);
+    }
+}

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);
+    }
+}

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);
+    }
+}

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);
+    }
+}

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);
+    }
+}

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);