diff --git a/build.gradle b/build.gradle index 1787987133dde..84b918c37e611 100644 --- a/build.gradle +++ b/build.gradle @@ -2777,9 +2777,11 @@ project(':connect:runtime') { testImplementation project(':clients').sourceSets.test.output testImplementation project(':core') testImplementation project(':metadata') + testImplementation project(':server-common') testImplementation project(':core').sourceSets.test.output testImplementation project(':server-common') testImplementation project(':connect:test-plugins') + testImplementation project(':server-common').sourceSets.test.output testImplementation libs.easymock testImplementation libs.junitJupiterApi @@ -2986,6 +2988,7 @@ project(':connect:mirror') { testImplementation project(':connect:runtime').sourceSets.test.output testImplementation project(':core') testImplementation project(':core').sourceSets.test.output + testImplementation project(':server-common').sourceSets.test.output testRuntimeOnly project(':connect:runtime') testRuntimeOnly libs.slf4jlog4j diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml index 44673899772c5..b78605ab5e61e 100644 --- a/checkstyle/import-control.xml +++ b/checkstyle/import-control.xml @@ -556,6 +556,7 @@ + @@ -574,6 +575,7 @@ + diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java index 35fa4c386657e..6bbd91164d3e7 100644 --- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java +++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java @@ -113,9 +113,6 @@ public void testSingleNodeCluster() throws Exception { EmbeddedKafkaCluster clusterA = startKafkaCluster("A", 1, brokerProps); EmbeddedKafkaCluster clusterB = startKafkaCluster("B", 1, brokerProps); - clusterA.start(); - clusterB.start(); - try (Admin adminB = clusterB.createAdminClient()) { // Cluster aliases @@ -187,9 +184,6 @@ public void testMultiNodeCluster() throws Exception { EmbeddedKafkaCluster clusterA = startKafkaCluster("A", 1, brokerProps); EmbeddedKafkaCluster clusterB = startKafkaCluster("B", 1, brokerProps); - clusterA.start(); - clusterB.start(); - try (Admin adminB = clusterB.createAdminClient()) { // Cluster aliases final String a = "A"; diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java index 4beeaaf876a74..e12e08eccf3ce 100644 --- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java +++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java @@ -30,45 +30,49 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.config.TopicConfig; import org.apache.kafka.common.utils.Exit; -import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Timer; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.connect.connector.Connector; +import org.apache.kafka.connect.mirror.Checkpoint; import org.apache.kafka.connect.mirror.DefaultConfigPropertyFilter; +import org.apache.kafka.connect.mirror.MirrorCheckpointConnector; import org.apache.kafka.connect.mirror.MirrorClient; import org.apache.kafka.connect.mirror.MirrorHeartbeatConnector; import org.apache.kafka.connect.mirror.MirrorMakerConfig; import org.apache.kafka.connect.mirror.MirrorSourceConnector; import org.apache.kafka.connect.mirror.MirrorUtils; import org.apache.kafka.connect.mirror.SourceAndTarget; -import org.apache.kafka.connect.mirror.Checkpoint; -import org.apache.kafka.connect.mirror.MirrorCheckpointConnector; import org.apache.kafka.connect.mirror.TestUtils; import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffset; import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets; import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster; import org.apache.kafka.connect.util.clusters.EmbeddedKafkaCluster; import org.apache.kafka.connect.util.clusters.UngracefulShutdownException; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; -import java.util.HashSet; -import java.util.List; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Properties; import java.util.Set; import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; @@ -77,19 +81,12 @@ import java.util.function.LongUnaryOperator; import java.util.stream.Collectors; -import org.junit.jupiter.api.Tag; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import static org.apache.kafka.test.TestUtils.waitForCondition; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; +import static org.junit.jupiter.api.Assertions.assertTrue; /** * Tests MM2 replication and failover/failback logic. @@ -429,12 +426,12 @@ public void testReplicationWithEmptyPartition() throws Exception { try (Consumer primaryConsumer = primary.kafka().createConsumerAndSubscribeTo(consumerProps, topic)) { waitForConsumingAllRecords(primaryConsumer, expectedRecords); } - + // one way replication from primary to backup mm2Props.put(BACKUP_CLUSTER_ALIAS + "->" + PRIMARY_CLUSTER_ALIAS + ".enabled", "false"); mm2Config = new MirrorMakerConfig(mm2Props); waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS); - + // sleep few seconds to have MM2 finish replication so that "end" consumer will consume some record Thread.sleep(TimeUnit.SECONDS.toMillis(3)); @@ -445,7 +442,7 @@ public void testReplicationWithEmptyPartition() throws Exception { backupTopic)) { waitForConsumingAllRecords(backupConsumer, expectedRecords); } - + try (Admin backupClient = backup.kafka().createAdminClient()) { // retrieve the consumer group offset from backup cluster Map remoteOffsets = @@ -1093,14 +1090,11 @@ protected Producer initializeProducer(EmbeddedConnectCluster clu * @param records Records to send in one parallel batch */ protected void produceMessages(Producer producer, List> records) { - List> futures = new ArrayList<>(); - for (ProducerRecord record : records) { - futures.add(producer.send(record)); - } Timer timer = Time.SYSTEM.timer(RECORD_PRODUCE_DURATION_MS); + try { - for (Future future : futures) { - future.get(timer.remainingMs(), TimeUnit.MILLISECONDS); + for (ProducerRecord record : records) { + producer.send(record).get(timer.remainingMs(), TimeUnit.MILLISECONDS); timer.update(); } } catch (ExecutionException | InterruptedException | TimeoutException e) { diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java index eb2af4842b60a..8a6f99d6ebe90 100644 --- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java +++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java @@ -16,10 +16,6 @@ */ package org.apache.kafka.connect.mirror.integration; -import java.util.Map; -import java.util.Properties; -import java.util.stream.Collectors; - import kafka.server.KafkaConfig; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.common.config.SslConfigs; @@ -27,10 +23,13 @@ import org.apache.kafka.common.network.Mode; import org.apache.kafka.test.TestSslUtils; import org.apache.kafka.test.TestUtils; - import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Tag; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; + /** * Tests MM2 replication with SSL enabled at backup kafka cluster */ @@ -41,25 +40,24 @@ public class MirrorConnectorsIntegrationSSLTest extends MirrorConnectorsIntegrat public void startClusters() throws Exception { Map sslConfig = TestSslUtils.createSslConfig(false, true, Mode.SERVER, TestUtils.tempFile(), "testCert"); // enable SSL on backup kafka broker - backupBrokerProps.put(KafkaConfig.ListenersProp(), "SSL://localhost:0"); - backupBrokerProps.put(KafkaConfig.InterBrokerListenerNameProp(), "SSL"); + backupBrokerProps.put(KafkaConfig.ListenerSecurityProtocolMapProp(), "EXTERNAL:SSL,CONTROLLER:SSL"); backupBrokerProps.putAll(sslConfig); - + Properties sslProps = new Properties(); sslProps.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sslConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG)); sslProps.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, ((Password) sslConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)).value()); sslProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL"); - + // set SSL config for kafka connect worker backupWorkerProps.putAll(sslProps.entrySet().stream().collect(Collectors.toMap( e -> String.valueOf(e.getKey()), e -> String.valueOf(e.getValue())))); - + mm2Props.putAll(sslProps.entrySet().stream().collect(Collectors.toMap( e -> BACKUP_CLUSTER_ALIAS + "." + e.getKey(), e -> String.valueOf(e.getValue())))); // set SSL config for producer used by source task in MM2 mm2Props.putAll(sslProps.entrySet().stream().collect(Collectors.toMap( e -> BACKUP_CLUSTER_ALIAS + ".producer." + e.getKey(), e -> String.valueOf(e.getValue())))); - + super.startClusters(); } } diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsWithCustomForwardingAdminIntegrationTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsWithCustomForwardingAdminIntegrationTest.java index 281bb88456e29..ccca8b1ac6798 100644 --- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsWithCustomForwardingAdminIntegrationTest.java +++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsWithCustomForwardingAdminIntegrationTest.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.connect.mirror.integration; +import kafka.server.KafkaConfig; import org.apache.kafka.clients.admin.Admin; import org.apache.kafka.clients.admin.NewPartitions; import org.apache.kafka.common.acl.AccessControlEntry; @@ -67,17 +68,18 @@ public class MirrorConnectorsWithCustomForwardingAdminIntegrationTest extends Mi * enable ACL on brokers. */ protected static void enableAclAuthorizer(Properties brokerProps) { - brokerProps.put("authorizer.class.name", "kafka.security.authorizer.AclAuthorizer"); - brokerProps.put("sasl.enabled.mechanisms", "PLAIN"); - brokerProps.put("sasl.mechanism.inter.broker.protocol", "PLAIN"); - brokerProps.put("security.inter.broker.protocol", "SASL_PLAINTEXT"); - brokerProps.put("listeners", "SASL_PLAINTEXT://localhost:0"); - brokerProps.put("listener.name.sasl_plaintext.plain.sasl.jaas.config", - "org.apache.kafka.common.security.plain.PlainLoginModule required " - + "username=\"super\" " - + "password=\"super_pwd\" " - + "user_connector=\"connector_pwd\" " - + "user_super=\"super_pwd\";"); + brokerProps.put(KafkaConfig.ListenerSecurityProtocolMapProp(), "CONTROLLER:SASL_PLAINTEXT,EXTERNAL:SASL_PLAINTEXT"); + brokerProps.put(KafkaConfig.AuthorizerClassNameProp(), "org.apache.kafka.metadata.authorizer.StandardAuthorizer"); + brokerProps.put(KafkaConfig.SaslEnabledMechanismsProp(), "PLAIN"); + brokerProps.put(KafkaConfig.SaslMechanismInterBrokerProtocolProp(), "PLAIN"); + brokerProps.put(KafkaConfig.SaslMechanismControllerProtocolProp(), "PLAIN"); + String listenerSaslJaasConfig = "org.apache.kafka.common.security.plain.PlainLoginModule required " + + "username=\"super\" " + + "password=\"super_pwd\" " + + "user_connector=\"connector_pwd\" " + + "user_super=\"super_pwd\";"; + brokerProps.put("listener.name.external.plain.sasl.jaas.config", listenerSaslJaasConfig); + brokerProps.put("listener.name.controller.plain.sasl.jaas.config", listenerSaslJaasConfig); brokerProps.put("super.users", "User:super"); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/BlockingConnectorTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/BlockingConnectorTest.java index 33614e317c490..714599896bf20 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/BlockingConnectorTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/BlockingConnectorTest.java @@ -117,7 +117,7 @@ public class BlockingConnectorTest { @Before public void setup() throws Exception { - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster connect = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) @@ -137,7 +137,7 @@ public void setup() throws Exception { @After public void close() { - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); Block.resetBlockLatch(); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java index 2e843cd6ec6a0..800445126244a 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.connect.integration; +import kafka.server.KafkaConfig; import org.apache.kafka.connect.runtime.distributed.DistributedConfig; import org.apache.kafka.connect.storage.StringConverter; import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster; @@ -30,6 +31,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.net.ServerSocket; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -89,7 +91,7 @@ public void setup() { brokerProps = new Properties(); brokerProps.put("auto.create.topics.enable", String.valueOf(false)); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster connectBuilder = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) @@ -100,7 +102,7 @@ public void setup() { @After public void close() { - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); } @@ -196,7 +198,20 @@ public void testRestartFailedTask() throws Exception { public void testBrokerCoordinator() throws Exception { ConnectorHandle connectorHandle = RuntimeHandles.get().connectorHandle(CONNECTOR_NAME); workerProps.put(DistributedConfig.SCHEDULED_REBALANCE_MAX_DELAY_MS_CONFIG, String.valueOf(5000)); - connect = connectBuilder.workerProps(workerProps).build(); + Properties brokerProps = new Properties(); + + // Find a free port and use it in the Kafka broker's listeners config. We can't use port 0 in the listeners + // config to get a random free port because in this test we want to stop the Kafka broker and then bring it + // back up and listening on the same port in order to verify that the Connect cluster can re-connect to Kafka + // and continue functioning normally. If we were to use port 0 here, the Kafka broker would most likely listen + // on a different random free port the second time it is started. Note that we can only use the static port + // because we have a single broker setup in this test. + int listenerPort; + try (ServerSocket s = new ServerSocket(0)) { + listenerPort = s.getLocalPort(); + } + brokerProps.put(KafkaConfig.ListenersProp(), String.format("EXTERNAL://localhost:%d,CONTROLLER://localhost:0", listenerPort)); + connect = connectBuilder.workerProps(workerProps).brokerProps(brokerProps).build(); // start the clusters connect.start(); int numTasks = 4; @@ -218,7 +233,7 @@ public void testBrokerCoordinator() throws Exception { // expect that the connector will be stopped once the coordinator is detected to be down StartAndStopLatch stopLatch = connectorHandle.expectedStops(1, false); - connect.kafka().stopOnlyKafka(); + connect.kafka().stopOnlyBrokers(); connect.assertions().assertExactlyNumWorkersAreUp(NUM_WORKERS, "Group of workers did not remain the same after broker shutdown"); @@ -233,7 +248,7 @@ public void testBrokerCoordinator() throws Exception { stopLatch.await(CONNECTOR_SETUP_DURATION_MS, TimeUnit.MILLISECONDS)); StartAndStopLatch startLatch = connectorHandle.expectedStarts(1, false); - connect.kafka().startOnlyKafkaOnSamePorts(); + connect.kafka().restartOnlyBrokers(); // Allow for the kafka brokers to come back online Thread.sleep(TimeUnit.SECONDS.toMillis(10)); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorClientPolicyIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorClientPolicyIntegrationTest.java index a0abece17d750..f210dc3a111d6 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorClientPolicyIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorClientPolicyIntegrationTest.java @@ -110,7 +110,7 @@ private EmbeddedConnectCluster connectClusterWithPolicy(String policy) throws In Properties exampleBrokerProps = new Properties(); exampleBrokerProps.put("auto.create.topics.enable", "false"); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster EmbeddedConnectCluster connect = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorRestartApiIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorRestartApiIntegrationTest.java index e399eee2ae93f..4d096a5f3a40e 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorRestartApiIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorRestartApiIntegrationTest.java @@ -124,7 +124,7 @@ public void tearDown() { @AfterClass public static void close() { - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connectClusterMap.values().forEach(EmbeddedConnectCluster::stop); } @@ -132,7 +132,7 @@ public static void close() { public void testRestartUnknownConnectorNoParams() throws Exception { String connectorName = "Unknown"; - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster startOrReuseConnectWithNumWorkers(ONE_WORKER); // Call the Restart API String restartEndpoint = connect.endpointForResource( @@ -153,7 +153,7 @@ public void testRestartUnknownConnector() throws Exception { private void restartUnknownConnector(boolean onlyFailed, boolean includeTasks) throws Exception { String connectorName = "Unknown"; - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster startOrReuseConnectWithNumWorkers(ONE_WORKER); // Call the Restart API String restartEndpoint = connect.endpointForResource( @@ -304,7 +304,7 @@ private void failedConnectorRestart(boolean onlyFailed, boolean includeTasks, in // setup up props for the source connector Map props = defaultSourceConnectorProps(TOPIC_NAME); props.put("connector.start.inject.error", "true"); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster startOrReuseConnectWithNumWorkers(ONE_WORKER); // Try to start the connector and its single task. @@ -335,7 +335,7 @@ private void failedTasksRestart(boolean onlyFailed, boolean includeTasks, int ex // setup up props for the source connector Map props = defaultSourceConnectorProps(TOPIC_NAME); tasksToFail.forEach(taskId -> props.put("task-" + taskId + ".start.inject.error", "true")); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster startOrReuseConnectWithNumWorkers(ONE_WORKER); // Try to start the connector and its single task. diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorTopicsIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorTopicsIntegrationTest.java index a8b812f8c31e7..ab73a738150a3 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorTopicsIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorTopicsIntegrationTest.java @@ -90,7 +90,7 @@ public void setup() { // setup Kafka broker properties brokerProps.put("auto.create.topics.enable", String.valueOf(false)); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster connectBuilder = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) @@ -101,7 +101,7 @@ public void setup() { @After public void close() { - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java index 21cd734065365..f934af342afbf 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.connect.integration; +import kafka.server.KafkaConfig; import org.apache.kafka.clients.admin.Admin; import org.apache.kafka.clients.admin.NewTopic; import org.apache.kafka.clients.consumer.ConsumerConfig; @@ -138,7 +139,7 @@ public void setup() { brokerProps.put("transaction.state.log.replication.factor", "1"); brokerProps.put("transaction.state.log.min.isr", "1"); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by Kafka connectBuilder = new EmbeddedConnectCluster.Builder() .numWorkers(DEFAULT_NUM_WORKERS) .numBrokers(1) @@ -157,7 +158,7 @@ private void startConnect() { @After public void close() { try { - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); } finally { // Clear the handle for the connector. Fun fact: if you don't do this, your tests become quite flaky. @@ -624,17 +625,18 @@ public void testConnectorReconfiguration() throws Exception { */ @Test public void testTasksFailOnInabilityToFence() throws Exception { - brokerProps.put("authorizer.class.name", "kafka.security.authorizer.AclAuthorizer"); - brokerProps.put("sasl.enabled.mechanisms", "PLAIN"); - brokerProps.put("sasl.mechanism.inter.broker.protocol", "PLAIN"); - brokerProps.put("security.inter.broker.protocol", "SASL_PLAINTEXT"); - brokerProps.put("listeners", "SASL_PLAINTEXT://localhost:0"); - brokerProps.put("listener.name.sasl_plaintext.plain.sasl.jaas.config", - "org.apache.kafka.common.security.plain.PlainLoginModule required " - + "username=\"super\" " - + "password=\"super_pwd\" " - + "user_connector=\"connector_pwd\" " - + "user_super=\"super_pwd\";"); + brokerProps.put(KafkaConfig.ListenerSecurityProtocolMapProp(), "CONTROLLER:SASL_PLAINTEXT,EXTERNAL:SASL_PLAINTEXT"); + brokerProps.put(KafkaConfig.AuthorizerClassNameProp(), "org.apache.kafka.metadata.authorizer.StandardAuthorizer"); + brokerProps.put(KafkaConfig.SaslEnabledMechanismsProp(), "PLAIN"); + brokerProps.put(KafkaConfig.SaslMechanismInterBrokerProtocolProp(), "PLAIN"); + brokerProps.put(KafkaConfig.SaslMechanismControllerProtocolProp(), "PLAIN"); + String listenerSaslJaasConfig = "org.apache.kafka.common.security.plain.PlainLoginModule required " + + "username=\"super\" " + + "password=\"super_pwd\" " + + "user_connector=\"connector_pwd\" " + + "user_super=\"super_pwd\";"; + brokerProps.put("listener.name.external.plain.sasl.jaas.config", listenerSaslJaasConfig); + brokerProps.put("listener.name.controller.plain.sasl.jaas.config", listenerSaslJaasConfig); brokerProps.put("super.users", "User:super"); Map superUserClientConfig = new HashMap<>(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExampleConnectIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExampleConnectIntegrationTest.java index 23a87c2c0fe49..f300afcdc9a40 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExampleConnectIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExampleConnectIntegrationTest.java @@ -84,7 +84,7 @@ public void setup() { Properties exampleBrokerProps = new Properties(); exampleBrokerProps.put("auto.create.topics.enable", "false"); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster connect = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) @@ -105,7 +105,7 @@ public void close() { // delete connector handle RuntimeHandles.get().deleteConnector(CONNECTOR_NAME); - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java index d73d1c4ed069e..d6222be85821c 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java @@ -55,7 +55,7 @@ public void setup() { @After public void close() { - // stop all Connect, Kafka and Zk threads. + // stop the Connect workers and Kafka brokers. connect.stop(); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java index 04e12ea41e01f..a933227033515 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java @@ -89,7 +89,7 @@ public void setup() { Properties brokerProps = new Properties(); brokerProps.put("auto.create.topics.enable", "false"); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster connect = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) @@ -104,7 +104,7 @@ public void setup() { @After public void close() { - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RestExtensionIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RestExtensionIntegrationTest.java index a0f993f2f63a8..218535aaf3adb 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RestExtensionIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RestExtensionIntegrationTest.java @@ -66,7 +66,7 @@ public void testRestExtensionApi() throws InterruptedException { Map workerProps = new HashMap<>(); workerProps.put(REST_EXTENSION_CLASSES_CONFIG, IntegrationTestRestExtension.class.getName()); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster connect = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) @@ -137,7 +137,7 @@ public void testRestExtensionApi() throws InterruptedException { @After public void close() { - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); IntegrationTestRestExtension.instance = null; } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SessionedProtocolIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SessionedProtocolIntegrationTest.java index 8956a86e7c73c..72a910869ee70 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SessionedProtocolIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SessionedProtocolIntegrationTest.java @@ -65,7 +65,7 @@ public void setup() { Map workerProps = new HashMap<>(); workerProps.put(CONNECT_PROTOCOL_CONFIG, ConnectProtocolCompatibility.SESSIONED.protocol()); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster connect = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(2) @@ -82,7 +82,7 @@ public void setup() { @After public void close() { - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SinkConnectorsIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SinkConnectorsIntegrationTest.java index a8bfbb291ffa1..44d596f1d9b34 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SinkConnectorsIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SinkConnectorsIntegrationTest.java @@ -75,7 +75,7 @@ public void setup() throws Exception { brokerProps.put("auto.create.topics.enable", "false"); brokerProps.put("delete.topic.enable", "true"); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster connect = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) @@ -91,7 +91,7 @@ public void close() { // delete connector handle RuntimeHandles.get().deleteConnector(CONNECTOR_NAME); - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SourceConnectorsIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SourceConnectorsIntegrationTest.java index b35b072080257..143c304cc222e 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SourceConnectorsIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SourceConnectorsIntegrationTest.java @@ -80,7 +80,7 @@ public void setup() { // setup Kafka broker properties brokerProps.put("auto.create.topics.enable", String.valueOf(false)); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster connectBuilder = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) @@ -91,7 +91,7 @@ public void setup() { @After public void close() { - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java index 02d8c7f71b1ac..981e48ef2513e 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java @@ -84,7 +84,7 @@ public void setup() { // This is required because tests in this class also test per-connector topic creation with transformations brokerProps.put("auto.create.topics.enable", "false"); - // build a Connect cluster backed by Kafka and Zk + // build a Connect cluster backed by a Kafka KRaft cluster connect = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) @@ -105,7 +105,7 @@ public void close() { // delete connector handle RuntimeHandles.get().deleteConnector(CONNECTOR_NAME); - // stop all Connect, Kafka and Zk threads. + // stop the Connect cluster and its backing Kafka cluster. connect.stop(); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java index 4f0c4369f89ed..0f0bcf7b5f163 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java @@ -67,9 +67,11 @@ import static org.apache.kafka.connect.runtime.rest.RestServerConfig.LISTENERS_CONFIG; /** - * Start an embedded connect worker. Internally, this class will spin up a Kafka and Zk cluster, setup any tmp - * directories and clean up them on them. Methods on the same {@code EmbeddedConnectCluster} are - * not guaranteed to be thread-safe. + * Start an embedded Connect cluster that can be used for integration tests. Internally, this class also spins up a + * backing Kafka KRaft cluster for the Connect cluster leveraging {@link kafka.testkit.KafkaClusterTestKit}. Methods + * on the same {@code EmbeddedConnectCluster} are not guaranteed to be thread-safe. This class also provides various + * utility methods to perform actions on the Connect cluster such as connector creation, config validation, connector + * restarts, pause / resume, connector deletion etc. */ public class EmbeddedConnectCluster { @@ -137,7 +139,7 @@ private EmbeddedConnectCluster(String name, Map workerProps, int }; /** - * Start the connect cluster and the embedded Kafka and Zookeeper cluster. + * Start the Connect cluster and the embedded Kafka KRaft cluster. */ public void start() { if (maskExitProcedures) { @@ -154,7 +156,7 @@ public void start() { } /** - * Stop the connect cluster and the embedded Kafka and Zookeeper cluster. + * Stop the Connect cluster and the embedded Kafka KRaft cluster. * Clean up any temp directories created locally. * * @throws RuntimeException if Kafka brokers fail to stop @@ -164,11 +166,6 @@ public void stop() { connectCluster.forEach(this::stopWorker); try { kafkaCluster.stop(); - } catch (UngracefulShutdownException e) { - log.warn("Kafka did not shutdown gracefully"); - } catch (Exception e) { - log.error("Could not stop kafka", e); - throw new RuntimeException("Could not stop brokers", e); } finally { if (maskExitProcedures) { Exit.resetExitProcedure(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java index df8a2253531f6..79577c93ec450 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java @@ -16,12 +16,10 @@ */ package org.apache.kafka.connect.util.clusters; -import kafka.cluster.EndPoint; +import kafka.server.BrokerServer; import kafka.server.KafkaConfig; -import kafka.server.KafkaServer; -import kafka.utils.CoreUtils; -import kafka.utils.TestUtils; -import kafka.zk.EmbeddedZookeeper; +import kafka.testkit.KafkaClusterTestKit; +import kafka.testkit.TestKitNodes; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.admin.Admin; import org.apache.kafka.clients.admin.AdminClientConfig; @@ -48,18 +46,13 @@ import org.apache.kafka.common.config.types.Password; import org.apache.kafka.common.errors.InvalidReplicationFactorException; import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; -import org.apache.kafka.common.network.ListenerName; import org.apache.kafka.common.serialization.ByteArraySerializer; -import org.apache.kafka.common.utils.MockTime; -import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.connect.errors.ConnectException; import org.apache.kafka.metadata.BrokerState; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.nio.file.Files; import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; @@ -77,6 +70,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.function.Predicate; import java.util.stream.Collectors; @@ -91,92 +85,57 @@ import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG; /** - * Setup an embedded Kafka cluster with specified number of brokers and specified broker properties. To be used for - * integration tests. + * Setup an embedded Kafka KRaft cluster (using {@link kafka.testkit.KafkaClusterTestKit} internally) with the + * specified number of brokers and the specified broker properties. This can be used for integration tests and is + * typically used in conjunction with {@link EmbeddedConnectCluster}. Additional Kafka client properties can also be + * supplied if required. This class also provides various utility methods to easily create Kafka topics, produce data, + * consume data etc. */ public class EmbeddedKafkaCluster { private static final Logger log = LoggerFactory.getLogger(EmbeddedKafkaCluster.class); - private static final long DEFAULT_PRODUCE_SEND_DURATION_MS = TimeUnit.SECONDS.toMillis(120); + private static final long DEFAULT_PRODUCE_SEND_DURATION_MS = TimeUnit.SECONDS.toMillis(120); - // Kafka Config - private final KafkaServer[] brokers; + private final KafkaClusterTestKit cluster; private final Properties brokerConfig; - private final Time time = new MockTime(); - private final int[] currentBrokerPorts; - private final String[] currentBrokerLogDirs; - private final boolean hasListenerConfig; - - final Map clientConfigs; - - private EmbeddedZookeeper zookeeper = null; - private ListenerName listenerName = new ListenerName("PLAINTEXT"); + private final Map clientConfigs; private KafkaProducer producer; - public EmbeddedKafkaCluster(final int numBrokers, - final Properties brokerConfig) { + public EmbeddedKafkaCluster(final int numBrokers, final Properties brokerConfig) { this(numBrokers, brokerConfig, Collections.emptyMap()); } public EmbeddedKafkaCluster(final int numBrokers, - final Properties brokerConfig, - final Map clientConfigs) { - brokers = new KafkaServer[numBrokers]; - currentBrokerPorts = new int[numBrokers]; - currentBrokerLogDirs = new String[numBrokers]; + final Properties brokerConfig, + final Map clientConfigs) { + addDefaultBrokerPropsIfAbsent(brokerConfig, numBrokers); + try { + KafkaClusterTestKit.Builder clusterBuilder = new KafkaClusterTestKit.Builder( + new TestKitNodes.Builder() + .setCombined(true) + .setNumBrokerNodes(numBrokers) + .setNumControllerNodes(numBrokers) + .build() + ); + + brokerConfig.forEach((k, v) -> clusterBuilder.setConfigProp((String) k, v)); + cluster = clusterBuilder.build(); + cluster.nonFatalFaultHandler().setIgnore(true); + } catch (Exception e) { + throw new ConnectException("Failed to create test Kafka cluster", e); + } this.brokerConfig = brokerConfig; - // Since we support `stop` followed by `startOnlyKafkaOnSamePorts`, we track whether - // a listener config is defined during initialization in order to know if it's - // safe to override it - hasListenerConfig = brokerConfig.get(KafkaConfig.ListenersProp()) != null; - this.clientConfigs = clientConfigs; } - /** - * Starts the Kafka cluster alone using the ports that were assigned during initialization of - * the harness. - * - * @throws ConnectException if a directory to store the data cannot be created - */ - public void startOnlyKafkaOnSamePorts() { - doStart(); - } - public void start() { - // pick a random port - zookeeper = new EmbeddedZookeeper(); - Arrays.fill(currentBrokerPorts, 0); - Arrays.fill(currentBrokerLogDirs, null); - doStart(); - } - - private void doStart() { - brokerConfig.put(KafkaConfig.ZkConnectProp(), zKConnectString()); - - putIfAbsent(brokerConfig, KafkaConfig.DeleteTopicEnableProp(), true); - putIfAbsent(brokerConfig, KafkaConfig.GroupInitialRebalanceDelayMsProp(), 0); - putIfAbsent(brokerConfig, KafkaConfig.OffsetsTopicReplicationFactorProp(), (short) brokers.length); - putIfAbsent(brokerConfig, KafkaConfig.AutoCreateTopicsEnableProp(), false); - // reduce the size of the log cleaner map to reduce test memory usage - putIfAbsent(brokerConfig, KafkaConfig.LogCleanerDedupeBufferSizeProp(), 2 * 1024 * 1024L); - - Object listenerConfig = brokerConfig.get(KafkaConfig.InterBrokerListenerNameProp()); - if (listenerConfig == null) - listenerConfig = brokerConfig.get(KafkaConfig.InterBrokerSecurityProtocolProp()); - if (listenerConfig == null) - listenerConfig = "PLAINTEXT"; - listenerName = new ListenerName(listenerConfig.toString()); - - for (int i = 0; i < brokers.length; i++) { - brokerConfig.put(KafkaConfig.BrokerIdProp(), i); - currentBrokerLogDirs[i] = currentBrokerLogDirs[i] == null ? createLogDir() : currentBrokerLogDirs[i]; - brokerConfig.put(KafkaConfig.LogDirProp(), currentBrokerLogDirs[i]); - if (!hasListenerConfig) - brokerConfig.put(KafkaConfig.ListenersProp(), listenerName.value() + "://localhost:" + currentBrokerPorts[i]); - brokers[i] = TestUtils.createServer(new KafkaConfig(brokerConfig, true), time); - currentBrokerPorts[i] = brokers[i].boundPort(listenerName); + try { + cluster.format(); + cluster.startup(); + cluster.waitForReadyBrokers(); + } catch (Exception e) { + throw new ConnectException("Failed to start test Kafka cluster", e); } Map producerProps = new HashMap<>(clientConfigs); @@ -189,112 +148,61 @@ private void doStart() { producer = new KafkaProducer<>(producerProps, new ByteArraySerializer(), new ByteArraySerializer()); } - public void stopOnlyKafka() { - stop(false, false); - } - - public void stop() { - stop(true, true); - } - - private void stop(boolean deleteLogDirs, boolean stopZK) { - try { - if (producer != null) { - producer.close(); - } - } catch (Exception e) { - log.error("Could not shutdown producer ", e); - throw new RuntimeException("Could not shutdown producer", e); - } - - for (KafkaServer broker : brokers) { - try { - broker.shutdown(); - } catch (Throwable t) { - String msg = String.format("Could not shutdown broker at %s", address(broker)); - log.error(msg, t); - throw new RuntimeException(msg, t); - } - } - - if (deleteLogDirs) { - for (KafkaServer broker : brokers) { - try { - log.info("Cleaning up kafka log dirs at {}", broker.config().logDirs()); - CoreUtils.delete(broker.config().logDirs()); - } catch (Throwable t) { - String msg = String.format("Could not clean up log dirs for broker at %s", - address(broker)); - log.error(msg, t); - throw new RuntimeException(msg, t); - } - } - } - - try { - if (stopZK) { - zookeeper.shutdown(); - } - } catch (Throwable t) { - String msg = String.format("Could not shutdown zookeeper at %s", zKConnectString()); - log.error(msg, t); - throw new RuntimeException(msg, t); - } + /** + * Restarts the Kafka brokers. This can be called after {@link #stopOnlyBrokers()}. Note that if the Kafka brokers + * need to be listening on the same ports as earlier, the {@link #brokerConfig} should contain the + * {@link KafkaConfig#ListenersProp} property and it should use a fixed non-zero free port. Also note that this is + * only possible when {@code numBrokers} is 1. + */ + public void restartOnlyBrokers() { + cluster.brokers().values().forEach(BrokerServer::startup); } - private static void putIfAbsent(final Properties props, final String propertyKey, final Object propertyValue) { - if (!props.containsKey(propertyKey)) { - props.put(propertyKey, propertyValue); - } + /** + * Stop only the Kafka brokers (and not the KRaft controllers). This can be used to test Connect's functionality + * when the backing Kafka cluster goes offline. + */ + public void stopOnlyBrokers() { + cluster.brokers().values().forEach(BrokerServer::shutdown); + cluster.brokers().values().forEach(BrokerServer::awaitShutdown); } - private String createLogDir() { - try { - return Files.createTempDirectory(getClass().getSimpleName()).toString(); - } catch (IOException e) { - log.error("Unable to create temporary log directory", e); - throw new ConnectException("Unable to create temporary log directory", e); + public void stop() { + AtomicReference shutdownFailure = new AtomicReference<>(); + Utils.closeQuietly(producer, "producer for embedded Kafka cluster", shutdownFailure); + Utils.closeQuietly(cluster, "embedded Kafka cluster", shutdownFailure); + if (shutdownFailure.get() != null) { + throw new ConnectException("Failed to shut down producer / embedded Kafka cluster", shutdownFailure.get()); } } public String bootstrapServers() { - return Arrays.stream(brokers) - .map(this::address) - .collect(Collectors.joining(",")); - } - - public String address(KafkaServer server) { - final EndPoint endPoint = server.advertisedListeners().head(); - return endPoint.host() + ":" + endPoint.port(); - } - - public String zKConnectString() { - return "127.0.0.1:" + zookeeper.port(); + return cluster.bootstrapServers(); } /** * Get the brokers that have a {@link BrokerState#RUNNING} state. * - * @return the list of {@link KafkaServer} instances that are running; - * never null but possibly empty + * @return the set of {@link BrokerServer} instances that are running; + * never null but possibly empty */ - public Set runningBrokers() { + public Set runningBrokers() { return brokersInState(state -> state == BrokerState.RUNNING); } /** * Get the brokers whose state match the given predicate. * - * @return the list of {@link KafkaServer} instances with states that match the predicate; - * never null but possibly empty + * @return the set of {@link BrokerServer} instances with states that match the predicate; + * never null but possibly empty */ - public Set brokersInState(Predicate desiredState) { - return Arrays.stream(brokers) - .filter(b -> hasState(b, desiredState)) - .collect(Collectors.toSet()); + public Set brokersInState(Predicate desiredState) { + return cluster.brokers().values().stream() + .filter(b -> hasState(b, desiredState)) + .collect(Collectors.toSet()); } - protected boolean hasState(KafkaServer server, Predicate desiredState) { + protected boolean hasState(BrokerServer server, Predicate desiredState) { try { return desiredState.test(server.brokerState()); } catch (Throwable e) { @@ -302,10 +210,10 @@ protected boolean hasState(KafkaServer server, Predicate desiredSta return false; } } - + public boolean sslEnabled() { - final String listeners = brokerConfig.getProperty(KafkaConfig.ListenersProp()); - return listeners != null && listeners.contains("SSL"); + final String listenerSecurityProtocolMap = brokerConfig.getProperty(KafkaConfig.ListenerSecurityProtocolMapProp()); + return listenerSecurityProtocolMap != null && listenerSecurityProtocolMap.contains("SSL"); } /** @@ -411,9 +319,9 @@ public void createTopic(String topic, int partitions, int replication, Map topicConfig, Map adminClientConfig) { - if (replication > brokers.length) { + if (replication > cluster.brokers().size()) { throw new InvalidReplicationFactorException("Insufficient brokers (" - + brokers.length + ") for desired replication (" + replication + ")"); + + cluster.brokers().size() + ") for desired replication (" + replication + ")"); } log.info("Creating topic { name: {}, partitions: {}, replication: {}, config: {} }", @@ -462,8 +370,7 @@ public Admin createAdminClient(Map adminClientConfig) { Properties props = Utils.mkProperties(clientConfigs); props.putAll(adminClientConfig); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers()); - final Object listeners = brokerConfig.get(KafkaConfig.ListenersProp()); - if (listeners != null && listeners.toString().contains("SSL")) { + if (sslEnabled()) { props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG)); props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, ((Password) brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)).value()); props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL"); @@ -635,16 +542,16 @@ public KafkaConsumer createConsumer(Map consumer Map props = new HashMap<>(clientConfigs); props.putAll(consumerProps); - putIfAbsent(props, GROUP_ID_CONFIG, UUID.randomUUID().toString()); - putIfAbsent(props, BOOTSTRAP_SERVERS_CONFIG, bootstrapServers()); - putIfAbsent(props, ENABLE_AUTO_COMMIT_CONFIG, "false"); - putIfAbsent(props, AUTO_OFFSET_RESET_CONFIG, "earliest"); - putIfAbsent(props, KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer"); - putIfAbsent(props, VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer"); + props.putIfAbsent(GROUP_ID_CONFIG, UUID.randomUUID().toString()); + props.putIfAbsent(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers()); + props.putIfAbsent(ENABLE_AUTO_COMMIT_CONFIG, "false"); + props.putIfAbsent(AUTO_OFFSET_RESET_CONFIG, "earliest"); + props.putIfAbsent(KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer"); + props.putIfAbsent(VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer"); if (sslEnabled()) { - putIfAbsent(props, SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG)); - putIfAbsent(props, SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)); - putIfAbsent(props, CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL"); + props.putIfAbsent(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG)); + props.putIfAbsent(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)); + props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL"); } KafkaConsumer consumer; try { @@ -664,13 +571,13 @@ public KafkaConsumer createConsumerAndSubscribeTo(Map createProducer(Map producerProps) { Map props = new HashMap<>(clientConfigs); props.putAll(producerProps); - putIfAbsent(props, BOOTSTRAP_SERVERS_CONFIG, bootstrapServers()); - putIfAbsent(props, KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); - putIfAbsent(props, VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); + props.putIfAbsent(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers()); + props.putIfAbsent(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); + props.putIfAbsent(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); if (sslEnabled()) { - putIfAbsent(props, SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG)); - putIfAbsent(props, SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)); - putIfAbsent(props, CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL"); + props.putIfAbsent(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG)); + props.putIfAbsent(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)); + props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL"); } KafkaProducer producer; try { @@ -681,9 +588,10 @@ public KafkaProducer createProducer(Map producer return producer; } - private static void putIfAbsent(final Map props, final String propertyKey, final Object propertyValue) { - if (!props.containsKey(propertyKey)) { - props.put(propertyKey, propertyValue); - } + private void addDefaultBrokerPropsIfAbsent(Properties brokerConfig, int numBrokers) { + brokerConfig.putIfAbsent(KafkaConfig.DeleteTopicEnableProp(), "true"); + brokerConfig.putIfAbsent(KafkaConfig.GroupInitialRebalanceDelayMsProp(), "0"); + brokerConfig.putIfAbsent(KafkaConfig.OffsetsTopicReplicationFactorProp(), String.valueOf(numBrokers)); + brokerConfig.putIfAbsent(KafkaConfig.AutoCreateTopicsEnableProp(), "false"); } } diff --git a/core/src/test/java/kafka/testkit/KafkaClusterTestKit.java b/core/src/test/java/kafka/testkit/KafkaClusterTestKit.java index e09ee49402f8c..e928650c19d4c 100644 --- a/core/src/test/java/kafka/testkit/KafkaClusterTestKit.java +++ b/core/src/test/java/kafka/testkit/KafkaClusterTestKit.java @@ -21,11 +21,11 @@ import kafka.server.BrokerServer; import kafka.server.ControllerServer; import kafka.server.FaultHandlerFactory; -import kafka.server.SharedServer; import kafka.server.KafkaConfig; import kafka.server.KafkaConfig$; import kafka.server.KafkaRaftServer; import kafka.server.MetaProperties; +import kafka.server.SharedServer; import kafka.tools.StorageTool; import kafka.utils.Logging; import org.apache.kafka.clients.CommonClientConfigs; @@ -139,14 +139,14 @@ public FaultHandler build(String name, boolean fatal, Runnable action) { public static class Builder { private TestKitNodes nodes; - private Map configProps = new HashMap<>(); - private SimpleFaultHandlerFactory faultHandlerFactory = new SimpleFaultHandlerFactory(); + private final Map configProps = new HashMap<>(); + private final SimpleFaultHandlerFactory faultHandlerFactory = new SimpleFaultHandlerFactory(); public Builder(TestKitNodes nodes) { this.nodes = nodes; } - public Builder setConfigProp(String key, String value) { + public Builder setConfigProp(String key, Object value) { this.configProps.put(key, value); return this; } @@ -155,7 +155,7 @@ private KafkaConfig createNodeConfig(TestKitNode node) { BrokerNode brokerNode = nodes.brokerNodes().get(node.id()); ControllerNode controllerNode = nodes.controllerNodes().get(node.id()); - Map props = new HashMap<>(configProps); + Map props = new HashMap<>(configProps); props.put(KafkaConfig$.MODULE$.ServerMaxStartupTimeMsProp(), Long.toString(TimeUnit.MINUTES.toMillis(10))); props.put(KafkaConfig$.MODULE$.ProcessRolesProp(), roles(node.id())); @@ -174,13 +174,16 @@ private KafkaConfig createNodeConfig(TestKitNode node) { props.put(KafkaConfig$.MODULE$.LogDirsProp(), String.join(",", brokerNode.logDataDirectories())); } - props.put(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(), - "EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT"); - props.put(KafkaConfig$.MODULE$.ListenersProp(), listeners(node.id())); - props.put(KafkaConfig$.MODULE$.InterBrokerListenerNameProp(), - nodes.interBrokerListenerName().value()); - props.put(KafkaConfig$.MODULE$.ControllerListenerNamesProp(), - "CONTROLLER"); + + // We allow configuring the listeners and related properties via Builder::setConfigProp, + // and they shouldn't be overridden here + props.putIfAbsent(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(), + "EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT"); + props.putIfAbsent(KafkaConfig$.MODULE$.ListenersProp(), listeners(node.id())); + props.putIfAbsent(KafkaConfig$.MODULE$.InterBrokerListenerNameProp(), + nodes.interBrokerListenerName().value()); + props.putIfAbsent(KafkaConfig$.MODULE$.ControllerListenerNamesProp(), "CONTROLLER"); + // Note: we can't accurately set controller.quorum.voters yet, since we don't // yet know what ports each controller will pick. Set it to a dummy string // for now as a placeholder. @@ -237,7 +240,7 @@ public KafkaClusterTestKit build() throws Exception { bootstrapMetadata); } catch (Throwable e) { log.error("Error creating controller {}", node.id(), e); - Utils.swallow(log, Level.WARN, "sharedServer.stopForController error", () -> sharedServer.stopForController()); + Utils.swallow(log, Level.WARN, "sharedServer.stopForController error", sharedServer::stopForController); if (controller != null) controller.shutdown(); throw e; } @@ -266,7 +269,7 @@ public KafkaClusterTestKit build() throws Exception { JavaConverters.asScalaBuffer(Collections.emptyList()).toSeq()); } catch (Throwable e) { log.error("Error creating broker {}", node.id(), e); - Utils.swallow(log, Level.WARN, "sharedServer.stopForBroker error", () -> sharedServer.stopForBroker()); + Utils.swallow(log, Level.WARN, "sharedServer.stopForBroker error", sharedServer::stopForBroker); if (broker != null) broker.shutdown(); throw e; } @@ -470,24 +473,28 @@ public Properties clientProperties() { public Properties clientProperties(Properties configOverrides) { if (!brokers.isEmpty()) { - StringBuilder bld = new StringBuilder(); - String prefix = ""; - for (Entry entry : brokers.entrySet()) { - int brokerId = entry.getKey(); - BrokerServer broker = entry.getValue(); - ListenerName listenerName = nodes.externalListenerName(); - int port = broker.boundPort(listenerName); - if (port <= 0) { - throw new RuntimeException("Broker " + brokerId + " does not yet " + + configOverrides.putIfAbsent(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers()); + } + return configOverrides; + } + + public String bootstrapServers() { + StringBuilder bld = new StringBuilder(); + String prefix = ""; + for (Entry entry : brokers.entrySet()) { + int brokerId = entry.getKey(); + BrokerServer broker = entry.getValue(); + ListenerName listenerName = broker.config().effectiveAdvertisedListeners().head().listenerName(); + int port = broker.boundPort(listenerName); + if (port <= 0) { + throw new RuntimeException("Broker " + brokerId + " does not yet " + "have a bound port for " + listenerName + ". Did you start " + "the cluster yet?"); - } - bld.append(prefix).append("localhost:").append(port); - prefix = ","; } - configOverrides.putIfAbsent(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bld.toString()); + bld.append(prefix).append("localhost:").append(port); + prefix = ","; } - return configOverrides; + return bld.toString(); } public Map controllers() {