[LI-HOTFIX] Add federated topic znode create rpc

checkstyle/suppressions.xml

@@ -162,6 +162,9 @@
     <suppress checks="MethodLength"
               files="KTableImpl.java"/>
 
+    <suppress checks="MethodLength"
+              files="(AbstractRequest|AbstractResponse).java"/>
+
     <suppress checks="ParameterNumber"
               files="StreamThread.java"/>
 

clients/src/main/java/org/apache/kafka/clients/admin/Admin.java

@@ -199,6 +199,13 @@ default CreateTopicsResult createTopics(Collection<NewTopic> newTopics) {
      */
     CreateTopicsResult createTopics(Collection<NewTopic> newTopics, CreateTopicsOptions options);
 
+    default CreateOrDeleteFederatedTopicZnodesResult createFederatedTopicZnodes(Map<String, String> federatedTopics) {
+        return createFederatedTopicZnodes(federatedTopics, new CreateFederatedTopicZnodesOptions());
+    }
+
+    CreateOrDeleteFederatedTopicZnodesResult createFederatedTopicZnodes(Map<String, String> federatedTopics,
+        CreateFederatedTopicZnodesOptions options);
+
     /**
      * This is a convenience method for {@link #deleteTopics(TopicCollection, DeleteTopicsOptions)}
      * with default options. See the overload for more details.

clients/src/main/java/org/apache/kafka/clients/admin/CreateFederatedTopicZnodesOptions.java

@@ -0,0 +1,58 @@
+/*
+ * 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.clients.admin;
+
+import java.util.Map;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+
+/**
+ * Options for {@link Admin#createFederatedTopicZnodes(Map, CreateFederatedTopicZnodesOptions)} (Collection)}.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class CreateFederatedTopicZnodesOptions extends AbstractOptions<CreateFederatedTopicZnodesOptions> {
+    private boolean retryOnQuotaViolation = true;
+
+    /**
+     * Set the timeout in milliseconds for this operation or {@code null} if the default api timeout for the
+     * AdminClient should be used.
+     *
+     */
+    // This method is retained to keep binary compatibility with 0.11
+    public CreateFederatedTopicZnodesOptions timeoutMs(Integer timeoutMs) {
+        this.timeoutMs = timeoutMs;
+        return this;
+    }
+
+    /**
+     * Set to true if quota violation should be automatically retried.
+     */
+    public CreateFederatedTopicZnodesOptions retryOnQuotaViolation(boolean retryOnQuotaViolation) {
+        this.retryOnQuotaViolation = retryOnQuotaViolation;
+        return this;
+    }
+
+    /**
+     * Returns true if quota violation should be automatically retried.
+     */
+    public boolean shouldRetryOnQuotaViolation() {
+        return retryOnQuotaViolation;
+    }
+}

clients/src/main/java/org/apache/kafka/clients/admin/CreateOrDeleteFederatedTopicZnodesResult.java

@@ -0,0 +1,40 @@
+/*
+ * 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.clients.admin;
+
+import java.util.Map;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+/**
+ * The result of {@link Admin#CreateFederatedTopicZnode()} or {@link Admin#DeleteFederatedTopicZnode()}.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class CreateOrDeleteFederatedTopicZnodesResult {
+    private final Map<String, KafkaFuture<Void>> future;
+
+    CreateOrDeleteFederatedTopicZnodesResult(Map<String, KafkaFuture<Void>> future) {
+        this.future = future;
+    }
+
+    public Map<String, KafkaFuture<Void>> all() {
+        return future;
+    }
+}

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -141,6 +141,7 @@
 import org.apache.kafka.common.message.ExpireDelegationTokenRequestData;
 import org.apache.kafka.common.message.LeaveGroupRequestData.MemberIdentity;
 import org.apache.kafka.common.message.LiControlledShutdownSkipSafetyCheckRequestData;
+import org.apache.kafka.common.message.LiCreateFederatedTopicZnodesRequestData;
 import org.apache.kafka.common.message.LiMoveControllerRequestData;
 import org.apache.kafka.common.message.ListGroupsRequestData;
 import org.apache.kafka.common.message.ListGroupsResponseData;
@@ -221,6 +222,8 @@
 import org.apache.kafka.common.requests.LiControlledShutdownSkipSafetyCheckResponse;
 import org.apache.kafka.common.requests.LiMoveControllerRequest;
 import org.apache.kafka.common.requests.LiMoveControllerResponse;
+import org.apache.kafka.common.requests.LiCreateFederatedTopicZnodesRequest;
+import org.apache.kafka.common.requests.LiCreateFederatedTopicZnodesResponse;
 import org.apache.kafka.common.requests.ListGroupsRequest;
 import org.apache.kafka.common.requests.ListGroupsResponse;
 import org.apache.kafka.common.requests.ListOffsetsRequest;
@@ -1596,6 +1599,45 @@ public CreateTopicsResult createTopics(final Collection<NewTopic> newTopics,
         return new CreateTopicsResult(new HashMap<>(topicFutures));
     }
 
+    @Override
+    public CreateOrDeleteFederatedTopicZnodesResult createFederatedTopicZnodes(final Map<String, String> federatedTopics,
+                                                                               final CreateFederatedTopicZnodesOptions options) {
+        final Map<String, KafkaFutureImpl<Void>> topicFutures = new HashMap<>(federatedTopics.size());
+        final long now = time.milliseconds();
+        List<LiCreateFederatedTopicZnodesRequestData.FederatedTopics> topics = new ArrayList<>();
+        federatedTopics.forEach((topic, namespace) -> {
+            topics.add(new LiCreateFederatedTopicZnodesRequestData.FederatedTopics().setName(topic).setNamespace(namespace));
+            topicFutures.put(topic, new KafkaFutureImpl<>());
+        });
+        runnable.call(new Call("createFederatedTopicZnodes", calcDeadlineMs(now, options.timeoutMs()),
+            new ControllerNodeProvider()) {
+            @Override
+            AbstractRequest.Builder<?> createRequest(int timeoutMs) {
+                return new LiCreateFederatedTopicZnodesRequest.Builder(new LiCreateFederatedTopicZnodesRequestData()
+                    .setTopics(topics)
+                    .setTimeoutMs(timeoutMs),  (short) 0);
+            }
+
+            @Override
+            void handleResponse(AbstractResponse abstractResponse) {
+                LiCreateFederatedTopicZnodesResponse response = (LiCreateFederatedTopicZnodesResponse) abstractResponse;
+                Errors errors = Errors.forCode(response.data().errorCode());
+                if (errors != Errors.NONE) {
+                    completeAllExceptionally(topicFutures.values(), errors.exception());
+                    return;
+                }
+                topicFutures.values().forEach(f -> f.complete(null));
+            }
+
+            @Override
+            void handleFailure(Throwable throwable) {
+                // Fail all the other remaining futures
+                completeAllExceptionally(topicFutures.values(), throwable);
+            }
+        }, now);
+        return new CreateOrDeleteFederatedTopicZnodesResult(new HashMap<>(topicFutures));
+    }
+
     private Call getCreateTopicsCall(final CreateTopicsOptions options,
                                      final Map<String, KafkaFutureImpl<TopicMetadataAndConfig>> futures,
                                      final CreatableTopicCollection topics,

clients/src/main/java/org/apache/kafka/clients/admin/NoOpAdminClient.java

@@ -46,6 +46,12 @@ public CreateTopicsResult createTopics(Collection<NewTopic> newTopics, CreateTop
         return null;
     }
 
+    @Override
+    public CreateOrDeleteFederatedTopicZnodesResult createFederatedTopicZnodes(Map<String, String> federatedTopics,
+        CreateFederatedTopicZnodesOptions options) {
+        return null;
+    }
+
     @Override
     public DeleteTopicsResult deleteTopics(TopicCollection topics, DeleteTopicsOptions options) {
         return null;

clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java

@@ -113,7 +113,9 @@ public enum ApiKeys {
     // LinkedIn API keys for APIs not yet upstreamed.
     LI_CONTROLLED_SHUTDOWN_SKIP_SAFETY_CHECK(ApiMessageType.LI_CONTROLLED_SHUTDOWN_SKIP_SAFETY_CHECK, true, true),
     LI_COMBINED_CONTROL(ApiMessageType.LI_COMBINED_CONTROL, true),
-    LI_MOVE_CONTROLLER(ApiMessageType.LI_MOVE_CONTROLLER, true);
+    LI_MOVE_CONTROLLER(ApiMessageType.LI_MOVE_CONTROLLER, true),
+
+    LI_CREATE_FEDERATED_TOPIC_ZNODES(ApiMessageType.LI_CREATE_FEDERATED_TOPIC_ZNODES, false, true);
 
     private static final Map<ApiMessageType.ListenerType, EnumSet<ApiKeys>> APIS_BY_LISTENER =
         new EnumMap<>(ApiMessageType.ListenerType.class);

clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java

@@ -309,6 +309,8 @@ private static AbstractRequest doParseRequest(ApiKeys apiKey, short apiVersion,
                 return LiCombinedControlRequest.parse(buffer, apiVersion);
             case LI_MOVE_CONTROLLER:
                 return LiMoveControllerRequest.parse(buffer, apiVersion);
+            case LI_CREATE_FEDERATED_TOPIC_ZNODES:
+                return LiCreateFederatedTopicZnodesRequest.parse(buffer, apiVersion);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseRequest`, the " +
                         "code should be updated to do so.", apiKey));

clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java

@@ -253,6 +253,8 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, ByteBuffer response
                 return LiCombinedControlResponse.parse(responseBuffer, version);
             case LI_MOVE_CONTROLLER:
                 return LiMoveControllerResponse.parse(responseBuffer, version);
+            case LI_CREATE_FEDERATED_TOPIC_ZNODES:
+                return LiCreateFederatedTopicZnodesResponse.parse(responseBuffer, version);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
                         "code should be updated to do so.", apiKey));

clients/src/main/java/org/apache/kafka/common/requests/LiCreateFederatedTopicZnodesRequest.java

@@ -0,0 +1,70 @@
+/*
+ * 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.requests;
+
+import java.nio.ByteBuffer;
+import org.apache.kafka.common.message.LiCreateFederatedTopicZnodesRequestData;
+import org.apache.kafka.common.message.LiCreateFederatedTopicZnodesResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.Errors;
+
+
+public class LiCreateFederatedTopicZnodesRequest extends AbstractRequest {
+    public static class Builder extends AbstractRequest.Builder<LiCreateFederatedTopicZnodesRequest> {
+        private final LiCreateFederatedTopicZnodesRequestData data;
+
+        public Builder(LiCreateFederatedTopicZnodesRequestData data, short allowedVersion) {
+            super(ApiKeys.LI_CREATE_FEDERATED_TOPIC_ZNODES, allowedVersion);
+            this.data = data;
+        }
+
+        @Override
+        public LiCreateFederatedTopicZnodesRequest build(short version) {
+            return new LiCreateFederatedTopicZnodesRequest(data, version);
+        }
+
+        @Override
+        public String toString() {
+            return data.toString();
+        }
+    }
+
+    private final LiCreateFederatedTopicZnodesRequestData data;
+
+    LiCreateFederatedTopicZnodesRequest(LiCreateFederatedTopicZnodesRequestData data, short version) {
+        super(ApiKeys.LI_CREATE_FEDERATED_TOPIC_ZNODES, version);
+        this.data = data;
+    }
+
+    public static LiCreateFederatedTopicZnodesRequest parse(ByteBuffer buffer, short version) {
+        return new LiCreateFederatedTopicZnodesRequest(new LiCreateFederatedTopicZnodesRequestData(new ByteBufferAccessor(buffer), version), version);
+    }
+
+    @Override
+    public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
+        LiCreateFederatedTopicZnodesResponseData data = new LiCreateFederatedTopicZnodesResponseData()
+            .setErrorCode(Errors.forException(e).code());
+        return new LiCreateFederatedTopicZnodesResponse(data, version());
+    }
+
+    @Override
+    public LiCreateFederatedTopicZnodesRequestData data() {
+        return data;
+    }
+}

clients/src/main/java/org/apache/kafka/common/requests/LiCreateFederatedTopicZnodesResponse.java

@@ -0,0 +1,77 @@
+/*
+ * 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.requests;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.kafka.common.message.LiCreateFederatedTopicZnodesResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.Errors;
+
+
+public class LiCreateFederatedTopicZnodesResponse extends AbstractResponse {
+    private final LiCreateFederatedTopicZnodesResponseData data;
+    private final short version;
+
+    public LiCreateFederatedTopicZnodesResponse(LiCreateFederatedTopicZnodesResponseData data, short version) {
+        super(ApiKeys.LI_CREATE_FEDERATED_TOPIC_ZNODES);
+        this.data = data;
+        this.version = version;
+    }
+
+    public Errors error() {
+        return Errors.forCode(data.errorCode());
+    }
+
+    @Override
+    public Map<Errors, Integer> errorCounts() {
+        return Collections.singletonMap(error(), 1);
+    }
+
+    @Override
+    public LiCreateFederatedTopicZnodesResponseData data() {
+        return data;
+    }
+
+    public static LiCreateFederatedTopicZnodesResponse prepareResponse(Errors error, int throttleTimeMs, short version) {
+        LiCreateFederatedTopicZnodesResponseData data = new LiCreateFederatedTopicZnodesResponseData();
+        data.setErrorCode(error.code());
+        data.setThrottleTimeMs(throttleTimeMs);
+        return new LiCreateFederatedTopicZnodesResponse(data, version);
+    }
+
+    public static LiCreateFederatedTopicZnodesResponse parse(ByteBuffer buffer, short version) {
+        return new LiCreateFederatedTopicZnodesResponse(new LiCreateFederatedTopicZnodesResponseData(new ByteBufferAccessor(buffer), version), version);
+    }
+
+    @Override
+    public int throttleTimeMs() {
+        return data.throttleTimeMs();
+    }
+
+    public short version() {
+        return version;
+    }
+
+    @Override
+    public String toString() {
+        return data.toString();
+    }
+}