[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572505265
##
File path: core/src/test/java/kafka/test/ClusterConfig.java
##
@@ -0,0 +1,207 @@
+/*
+ * 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 kafka.test;
+
+import kafka.test.annotation.Type;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+
+import java.io.File;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Represents a requested configuration of a Kafka cluster for integration
testing
+ */
+public class ClusterConfig {
+
+private final Type type;
+private final int brokers;
+private final int controllers;
+private final String name;
+private final boolean autoStart;
+
+private final String securityProtocol;
+private final String listenerName;
+private final File trustStoreFile;
+
+private final Properties serverProperties = new Properties();
+private final Properties producerProperties = new Properties();
+private final Properties consumerProperties = new Properties();
+private final Properties adminClientProperties = new Properties();
+private final Properties saslServerProperties = new Properties();
+private final Properties saslClientProperties = new Properties();
+
+ClusterConfig(Type type, int brokers, int controllers, String name,
boolean autoStart,
+ String securityProtocol, String listenerName, File
trustStoreFile) {
+this.type = type;
+this.brokers = brokers;
+this.controllers = controllers;
+this.name = name;
+this.autoStart = autoStart;
+this.securityProtocol = securityProtocol;
+this.listenerName = listenerName;
+this.trustStoreFile = trustStoreFile;
+}
+
+public Type clusterType() {
+return type;
+}
+
+public int brokers() {
+return brokers;
+}
+
+public int controllers() {
+return controllers;
Review comment:
ok
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572392149
##
File path: core/src/test/java/kafka/test/annotation/AutoStart.java
##
@@ -0,0 +1,24 @@
+/*
+ * 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 kafka.test.annotation;
+
+public enum AutoStart {
Review comment:
Thanks for the explanation. That makes sense.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572310383
##
File path:
core/src/test/scala/integration/kafka/server/IntegrationTestHelper.scala
##
@@ -0,0 +1,115 @@
+/*
+ * 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 integration.kafka.server
+
+import kafka.network.SocketServer
+import kafka.utils.{NotNothing, TestUtils}
+import org.apache.kafka.common.network.{ListenerName, Mode}
+import org.apache.kafka.common.protocol.ApiKeys
+import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse,
RequestHeader, RequestTestUtils, ResponseHeader}
+import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.common.utils.Utils
+
+import java.io.{DataInputStream, DataOutputStream}
+import java.net.Socket
+import java.nio.ByteBuffer
+import java.util.Properties
+import scala.annotation.nowarn
+import scala.reflect.ClassTag
+
+class IntegrationTestHelper {
Review comment:
These are all static methods, right? Can we just make this an "object"
rather than a class and have something like `object IntegrationTestUtils` ?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572309452
##
File path: core/src/test/java/kafka/test/junit/ZkClusterInvocationContext.java
##
@@ -0,0 +1,250 @@
+/*
+ * 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 kafka.test.junit;
+
+import integration.kafka.server.IntegrationTestHelper;
+import kafka.api.IntegrationTestHarness;
+import kafka.network.SocketServer;
+import kafka.server.KafkaServer;
+import kafka.utils.TestUtils;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import kafka.test.ClusterConfig;
+import kafka.test.ClusterInstance;
+import org.junit.jupiter.api.extension.AfterTestExecutionCallback;
+import org.junit.jupiter.api.extension.BeforeTestExecutionCallback;
+import org.junit.jupiter.api.extension.Extension;
+import org.junit.jupiter.api.extension.TestTemplateInvocationContext;
+import scala.Option;
+import scala.collection.JavaConverters;
+import scala.compat.java8.OptionConverters;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+/**
+ * Wraps a {@link IntegrationTestHarness} inside lifecycle methods for a test
invocation. Each instance of this
+ * class is provided with a configuration for the cluster.
+ *
+ * This context also provides parameter resolvers for:
+ *
+ *
+ * ClusterConfig (the same instance passed to the constructor)
+ * ClusterInstance (includes methods to expose underlying
SocketServer-s)
+ * IntegrationTestHelper (helper methods)
+ *
+ */
+public class ZkClusterInvocationContext implements
TestTemplateInvocationContext {
+
+private final ClusterConfig clusterConfig;
+private final AtomicReference clusterReference;
+
+public ZkClusterInvocationContext(ClusterConfig clusterConfig) {
+this.clusterConfig = clusterConfig;
+this.clusterReference = new AtomicReference<>();
+}
+
+@Override
+public String getDisplayName(int invocationIndex) {
+String clusterDesc = clusterConfig.nameTags().entrySet().stream()
+.map(Object::toString)
+.collect(Collectors.joining(", "));
+return String.format("[Zk %d] %s", invocationIndex, clusterDesc);
+}
+
+@Override
+public List getAdditionalExtensions() {
+ClusterInstance clusterShim = new ZkClusterInstance(clusterConfig,
clusterReference);
+return Arrays.asList(
+(BeforeTestExecutionCallback) context -> {
+// We have to wait to actually create the underlying cluster
until after our @BeforeEach methods
+// have run. This allows tests to set up external dependencies
like ZK, MiniKDC, etc.
+// However, since we cannot create this instance until we are
inside the test invocation, we have
+// to use a container class (AtomicReference) to provide this
cluster object to the test itself
+
+// This is what tests normally extend from to start a cluster,
here we create it anonymously and
+// configure the cluster using values from ClusterConfig
+IntegrationTestHarness cluster = new IntegrationTestHarness() {
+@Override
+public Properties serverConfig() {
+return clusterConfig.serverProperties();
+}
+
+@Override
+public Properties adminClientConfig() {
+return clusterConfig.adminClientProperties();
+}
+
+@Override
+public Properties consumerConfig() {
+return clusterConfig.consumerProperties();
+}
+
+@Override
+public Properties producerConfig() {
+return clusterConfig.producerProper
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572291568
##
File path: core/src/test/java/kafka/test/annotation/AutoStart.java
##
@@ -0,0 +1,24 @@
+/*
+ * 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 kafka.test.annotation;
+
+public enum AutoStart {
Review comment:
capitalize enum names please. also, do we need this enum? Why does
DEFAULT need to be distinct from YES? Once we have a lot of tests depending on
the default being "on" I can't see us changing this (or how having 3 values
would help us change it if so...)
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572290904
##
File path: core/src/test/java/kafka/test/ClusterInstance.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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 kafka.test;
+
+import kafka.network.SocketServer;
+import kafka.test.annotation.ClusterTest;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Properties;
+
+public interface ClusterInstance {
+
+enum ClusterType {
+Zk,
+// Raft
+}
+
+/**
+ * Cluster type. For now, only ZK is supported.
+ */
+ClusterType clusterType();
+
+/**
+ * The cluster configuration used to create this cluster. Changing data in
this instance through this accessor will
+ * have no affect on the cluster since it is already provisioned.
+ */
+ClusterConfig config();
+
+/**
+ * The listener for this cluster as configured by {@link ClusterTest} or
by {@link ClusterConfig}. If
+ * unspecified by those sources, this will return the listener for the
default security protocol PLAINTEXT
+ */
+ListenerName listener();
+
+/**
+ * The broker connect string which can be used by clients for bootstrapping
+ */
+String brokerList();
+
+/**
+ * A collection of all brokers in the cluster. In ZK-based clusters this
will also include the broker which is
+ * acting as the controller (since ZK controllers serve both broker and
controller roles).
+ */
+Collection brokers();
+
+/**
+ * A collection of all controllers in the cluster. For ZK-based clusters,
this will return the broker which is also
+ * currently the active controller. For Raft-based clusters, this will
return all controller servers.
+ */
+Collection controllers();
+
+/**
+ * Any one of the broker servers.
+ */
+Optional anyBroker();
+
+/**
+ * Any one of the controller servers.
+ */
+Optional anyController();
+
+/**
+ * The underlying object which is responsible for setting up and tearing
down the cluster.
+ */
+Object getUnderlying();
+
+default T getUnderlying(Class asClass) {
+return asClass.cast(getUnderlying());
+}
+
+Admin createAdminClient(Properties configOverrides);
+
+default Admin createAdminClient() {
+return createAdminClient(new Properties());
+}
+
+void start();
+
+void stop();
Review comment:
ok
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572290383
##
File path: core/src/test/java/kafka/test/ClusterInstance.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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 kafka.test;
+
+import kafka.network.SocketServer;
+import kafka.test.annotation.ClusterTest;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Properties;
+
+public interface ClusterInstance {
+
+enum ClusterType {
+ZK,
+// Raft
+}
+
+/**
+ * Cluster type. For now, only ZK is supported.
+ */
+ClusterType clusterType();
+
+/**
+ * The cluster configuration used to create this cluster. Changing data in
this instance through this accessor will
+ * have no affect on the cluster since it is already provisioned.
+ */
+ClusterConfig config();
+
+/**
+ * The listener for this cluster as configured by {@link ClusterTest} or
by {@link ClusterConfig}. If
+ * unspecified by those sources, this will return the listener for the
default security protocol PLAINTEXT
+ */
+ListenerName listener();
+
+/**
+ * The broker connect string which can be used by clients for bootstrapping
+ */
+String brokerList();
+
+/**
+ * A collection of all brokers in the cluster. In ZK-based clusters this
will also include the broker which is
+ * acting as the controller (since ZK controllers serve both broker and
controller roles).
+ */
+Collection brokerSocketServers();
+
+/**
+ * A collection of all controllers in the cluster. For ZK-based clusters,
this will return the broker which is also
+ * currently the active controller. For Raft-based clusters, this will
return all controller servers.
+ */
+Collection controllerSocketServers();
+
+/**
+ * Any one of the broker servers.
+ */
+Optional anyBrokerSocketServer();
+
+/**
+ * Any one of the controller servers.
+ */
+Optional anyControllerSocketServer();
Review comment:
Same question... can we just throw an exception if there are no
controller socket servers?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572290152
##
File path: core/src/test/java/kafka/test/ClusterInstance.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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 kafka.test;
+
+import kafka.network.SocketServer;
+import kafka.test.annotation.ClusterTest;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Properties;
+
+public interface ClusterInstance {
+
+enum ClusterType {
+ZK,
+// Raft
+}
+
+/**
+ * Cluster type. For now, only ZK is supported.
+ */
+ClusterType clusterType();
+
+/**
+ * The cluster configuration used to create this cluster. Changing data in
this instance through this accessor will
+ * have no affect on the cluster since it is already provisioned.
+ */
+ClusterConfig config();
+
+/**
+ * The listener for this cluster as configured by {@link ClusterTest} or
by {@link ClusterConfig}. If
+ * unspecified by those sources, this will return the listener for the
default security protocol PLAINTEXT
+ */
+ListenerName listener();
+
+/**
+ * The broker connect string which can be used by clients for bootstrapping
+ */
+String brokerList();
+
+/**
+ * A collection of all brokers in the cluster. In ZK-based clusters this
will also include the broker which is
+ * acting as the controller (since ZK controllers serve both broker and
controller roles).
+ */
+Collection brokerSocketServers();
+
+/**
+ * A collection of all controllers in the cluster. For ZK-based clusters,
this will return the broker which is also
+ * currently the active controller. For Raft-based clusters, this will
return all controller servers.
+ */
+Collection controllerSocketServers();
+
+/**
+ * Any one of the broker servers.
+ */
+Optional anyBrokerSocketServer();
Review comment:
Why does this return an `Optional<>`? Zero-node clusters seem like a
corner case we don't really care about (we can just throw an exception
there)
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572289080
##
File path: core/src/test/java/kafka/test/ClusterInstance.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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 kafka.test;
+
+import kafka.network.SocketServer;
+import kafka.test.annotation.ClusterTest;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Properties;
+
+public interface ClusterInstance {
+
+enum ClusterType {
+ZK,
+// Raft
+}
+
+/**
+ * Cluster type. For now, only ZK is supported.
+ */
+ClusterType clusterType();
+
+/**
+ * The cluster configuration used to create this cluster. Changing data in
this instance through this accessor will
+ * have no affect on the cluster since it is already provisioned.
+ */
+ClusterConfig config();
+
+/**
+ * The listener for this cluster as configured by {@link ClusterTest} or
by {@link ClusterConfig}. If
+ * unspecified by those sources, this will return the listener for the
default security protocol PLAINTEXT
+ */
+ListenerName listener();
+
+/**
+ * The broker connect string which can be used by clients for bootstrapping
+ */
+String brokerList();
Review comment:
I realize `brokerList` is the existing name, but given that we've
standardized on `--bootstrap-server` for the command-line argument, maybe it
makes sense to use that name here?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572286453
##
File path: core/src/test/java/kafka/test/ClusterInstance.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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 kafka.test;
+
+import kafka.network.SocketServer;
+import kafka.test.annotation.ClusterTest;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Properties;
+
+public interface ClusterInstance {
+
+enum ClusterType {
+Zk,
+// Raft
+}
+
+/**
+ * Cluster type. For now, only ZK is supported.
+ */
+ClusterType clusterType();
+
+/**
+ * The cluster configuration used to create this cluster. Changing data in
this instance through this accessor will
+ * have no affect on the cluster since it is already provisioned.
+ */
+ClusterConfig config();
+
+/**
+ * The listener for this cluster as configured by {@link ClusterTest} or
by {@link ClusterConfig}. If
+ * unspecified by those sources, this will return the listener for the
default security protocol PLAINTEXT
+ */
Review comment:
How about naming these "clientSecurityProtocol", "clientListenerName",
etc. to reflect the fact that the clients will be using these security
protocols, listener names, etc. Then we can add more listeners later for
specific tests without doing a big rename.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572284766
##
File path: core/src/test/java/kafka/test/ClusterConfig.java
##
@@ -0,0 +1,207 @@
+/*
+ * 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 kafka.test;
+
+import kafka.test.annotation.Type;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+
+import java.io.File;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Represents a requested configuration of a Kafka cluster for integration
testing
+ */
+public class ClusterConfig {
+
+private final Type type;
+private final int brokers;
+private final int controllers;
+private final String name;
+private final boolean autoStart;
+
+private final String securityProtocol;
+private final String listenerName;
+private final File trustStoreFile;
+
+private final Properties serverProperties = new Properties();
+private final Properties producerProperties = new Properties();
+private final Properties consumerProperties = new Properties();
+private final Properties adminClientProperties = new Properties();
+private final Properties saslServerProperties = new Properties();
+private final Properties saslClientProperties = new Properties();
+
+ClusterConfig(Type type, int brokers, int controllers, String name,
boolean autoStart,
+ String securityProtocol, String listenerName, File
trustStoreFile) {
+this.type = type;
+this.brokers = brokers;
+this.controllers = controllers;
+this.name = name;
+this.autoStart = autoStart;
+this.securityProtocol = securityProtocol;
+this.listenerName = listenerName;
+this.trustStoreFile = trustStoreFile;
+}
+
+public Type clusterType() {
+return type;
+}
+
+public int brokers() {
+return brokers;
+}
+
+public int controllers() {
+return controllers;
Review comment:
What does this return in ZK mode?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572284631
##
File path: core/src/test/java/kafka/test/ClusterConfig.java
##
@@ -0,0 +1,207 @@
+/*
+ * 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 kafka.test;
+
+import kafka.test.annotation.Type;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+
+import java.io.File;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Represents a requested configuration of a Kafka cluster for integration
testing
+ */
+public class ClusterConfig {
+
+private final Type type;
+private final int brokers;
+private final int controllers;
+private final String name;
+private final boolean autoStart;
+
+private final String securityProtocol;
+private final String listenerName;
+private final File trustStoreFile;
+
+private final Properties serverProperties = new Properties();
+private final Properties producerProperties = new Properties();
+private final Properties consumerProperties = new Properties();
+private final Properties adminClientProperties = new Properties();
+private final Properties saslServerProperties = new Properties();
+private final Properties saslClientProperties = new Properties();
+
+ClusterConfig(Type type, int brokers, int controllers, String name,
boolean autoStart,
+ String securityProtocol, String listenerName, File
trustStoreFile) {
+this.type = type;
+this.brokers = brokers;
+this.controllers = controllers;
+this.name = name;
+this.autoStart = autoStart;
+this.securityProtocol = securityProtocol;
+this.listenerName = listenerName;
+this.trustStoreFile = trustStoreFile;
+}
+
+public Type clusterType() {
+return type;
+}
+
+public int brokers() {
Review comment:
let's use "numBrokers" , etc. to reflect the fact that we're not
returning the actual brokers. Same for "controllers"
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572284053
##
File path: core/src/test/java/kafka/test/ClusterConfig.java
##
@@ -0,0 +1,207 @@
+/*
+ * 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 kafka.test;
+
+import kafka.test.annotation.Type;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+
+import java.io.File;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Represents a requested configuration of a Kafka cluster for integration
testing
+ */
+public class ClusterConfig {
+
+private final Type type;
+private final int brokers;
+private final int controllers;
+private final String name;
+private final boolean autoStart;
+
+private final String securityProtocol;
Review comment:
If there's no reason to use a string then let's just use the
`SecurityProtocol` enum.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r572284053
##
File path: core/src/test/java/kafka/test/ClusterConfig.java
##
@@ -0,0 +1,207 @@
+/*
+ * 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 kafka.test;
+
+import kafka.test.annotation.Type;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+
+import java.io.File;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Represents a requested configuration of a Kafka cluster for integration
testing
+ */
+public class ClusterConfig {
+
+private final Type type;
+private final int brokers;
+private final int controllers;
+private final String name;
+private final boolean autoStart;
+
+private final String securityProtocol;
Review comment:
If there's no reason to use a string then let's just use the
`SecurityProtocol` enum. There's no reason to create tech debt, this is all
new stuff.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r571246502
##
File path: core/src/test/java/kafka/test/ClusterInstance.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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 kafka.test;
+
+import kafka.network.SocketServer;
+import kafka.test.annotation.ClusterTest;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Properties;
+
+public interface ClusterInstance {
+
+enum ClusterType {
+Zk,
+// Raft
+}
+
+/**
+ * Cluster type. For now, only ZK is supported.
+ */
+ClusterType clusterType();
+
+/**
+ * The cluster configuration used to create this cluster. Changing data in
this instance through this accessor will
+ * have no affect on the cluster since it is already provisioned.
+ */
+ClusterConfig config();
+
+/**
+ * The listener for this cluster as configured by {@link ClusterTest} or
by {@link ClusterConfig}. If
+ * unspecified by those sources, this will return the listener for the
default security protocol PLAINTEXT
+ */
+ListenerName listener();
+
+/**
+ * The broker connect string which can be used by clients for bootstrapping
+ */
+String brokerList();
+
+/**
+ * A collection of all brokers in the cluster. In ZK-based clusters this
will also include the broker which is
+ * acting as the controller (since ZK controllers serve both broker and
controller roles).
+ */
+Collection brokers();
+
+/**
+ * A collection of all controllers in the cluster. For ZK-based clusters,
this will return the broker which is also
+ * currently the active controller. For Raft-based clusters, this will
return all controller servers.
+ */
+Collection controllers();
+
+/**
+ * Any one of the broker servers.
+ */
+Optional anyBroker();
+
+/**
+ * Any one of the controller servers.
+ */
+Optional anyController();
+
+/**
+ * The underlying object which is responsible for setting up and tearing
down the cluster.
+ */
+Object getUnderlying();
+
+default T getUnderlying(Class asClass) {
+return asClass.cast(getUnderlying());
+}
+
+Admin createAdminClient(Properties configOverrides);
+
+default Admin createAdminClient() {
+return createAdminClient(new Properties());
+}
+
+void start();
+
+void stop();
Review comment:
should this implement `AutoCloseable`? Does code need to remember to
close this cluster?
Or is that handled automatically by the framework...
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r571245971
##
File path: core/src/test/java/kafka/test/ClusterInstance.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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 kafka.test;
+
+import kafka.network.SocketServer;
+import kafka.test.annotation.ClusterTest;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Properties;
+
+public interface ClusterInstance {
+
+enum ClusterType {
+Zk,
+// Raft
+}
+
+/**
+ * Cluster type. For now, only ZK is supported.
+ */
+ClusterType clusterType();
+
+/**
+ * The cluster configuration used to create this cluster. Changing data in
this instance through this accessor will
+ * have no affect on the cluster since it is already provisioned.
+ */
+ClusterConfig config();
+
+/**
+ * The listener for this cluster as configured by {@link ClusterTest} or
by {@link ClusterConfig}. If
+ * unspecified by those sources, this will return the listener for the
default security protocol PLAINTEXT
+ */
+ListenerName listener();
+
+/**
+ * The broker connect string which can be used by clients for bootstrapping
+ */
+String brokerList();
+
+/**
+ * A collection of all brokers in the cluster. In ZK-based clusters this
will also include the broker which is
+ * acting as the controller (since ZK controllers serve both broker and
controller roles).
+ */
+Collection brokers();
Review comment:
I would prefer to call this method something like `brokerSocketServers`.
After all, the `SocketServer` is just one small part of the broker, not the
broker itself.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r571245403
##
File path: core/src/test/java/kafka/test/ClusterInstance.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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 kafka.test;
+
+import kafka.network.SocketServer;
+import kafka.test.annotation.ClusterTest;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Properties;
+
+public interface ClusterInstance {
+
+enum ClusterType {
+Zk,
+// Raft
+}
+
+/**
+ * Cluster type. For now, only ZK is supported.
+ */
+ClusterType clusterType();
+
+/**
+ * The cluster configuration used to create this cluster. Changing data in
this instance through this accessor will
+ * have no affect on the cluster since it is already provisioned.
+ */
+ClusterConfig config();
+
+/**
+ * The listener for this cluster as configured by {@link ClusterTest} or
by {@link ClusterConfig}. If
+ * unspecified by those sources, this will return the listener for the
default security protocol PLAINTEXT
+ */
Review comment:
A cluster can have multiple listeners, right? Is this intended to be
the listener that should be used by clients? If so, calling it the client
listener makes sense.
(Technically we could have multiple client listeners too, but that's much
less common)
Many clusters run with 3 listeners:
* client
* inter-broker NON-REPLICATION
* replication
and we are going to add a 4th:
* controller
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r571244427
##
File path: core/src/test/java/kafka/test/ClusterInstance.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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 kafka.test;
+
+import kafka.network.SocketServer;
+import kafka.test.annotation.ClusterTest;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Properties;
+
+public interface ClusterInstance {
+
+enum ClusterType {
Review comment:
Enum values should be capitalized
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r571242927
##
File path: core/src/test/java/kafka/test/ClusterConfig.java
##
@@ -0,0 +1,207 @@
+/*
+ * 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 kafka.test;
+
+import kafka.test.annotation.Type;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+
+import java.io.File;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Represents a requested configuration of a Kafka cluster for integration
testing
+ */
+public class ClusterConfig {
+
+private final Type type;
+private final int brokers;
+private final int controllers;
+private final String name;
+private final boolean autoStart;
+
+private final String securityProtocol;
Review comment:
What's the advantage of making this a string rather than using the
SecurityProtocol enum?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9986: JUnit extensions for integration tests
cmccabe commented on a change in pull request #9986:
URL: https://github.com/apache/kafka/pull/9986#discussion_r571241978
##
File path: core/src/test/java/kafka/test/annotation/Type.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 kafka.test.annotation;
+
+public enum Type {
Review comment:
Java Enums should be capitalized. Also can we add a JavaDoc for each
value describing what it is?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
