diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java index eb618fb0053f3..f3861ddd59bf9 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java @@ -1538,10 +1538,11 @@ public void onAssigned(ExtendedAssignment assignment, int generation) { short priorProtocolVersion = currentProtocolVersion; DistributedHerder.this.currentProtocolVersion = member.currentProtocolVersion(); log.info( - "Joined group at generation {} with protocol version {} and got assignment: {}", + "Joined group at generation {} with protocol version {} and got assignment: {} with rebalance delay: {}", generation, DistributedHerder.this.currentProtocolVersion, - assignment + assignment, + assignment.delay() ); synchronized (DistributedHerder.this) { DistributedHerder.this.assignment = assignment; @@ -1611,12 +1612,13 @@ public void onRevoked(String leader, Collection connectors, Collection performTaskAssignment(String leaderId, long maxOffset, Map memberConfigs, WorkerCoordinator coordinator, short protocolVersion) { + log.debug("Performing task assignment during generation: {} with memberId: {}", + coordinator.generationId(), coordinator.memberId()); + // Base set: The previous assignment of connectors-and-tasks is a standalone snapshot that // can be used to calculate derived sets log.debug("Previous assignments: {}", previousAssignment); @@ -350,6 +353,7 @@ protected void handleLostAssignments(ConnectorsAndTasks lostAssignments, ConnectorsAndTasks newSubmissions, List completeWorkerAssignment) { if (lostAssignments.isEmpty()) { + resetDelay(); return; } @@ -359,6 +363,7 @@ protected void handleLostAssignments(ConnectorsAndTasks lostAssignments, if (scheduledRebalance > 0 && now >= scheduledRebalance) { // delayed rebalance expired and it's time to assign resources + log.debug("Delayed rebalance expired. Reassigning lost tasks"); Optional candidateWorkerLoad = Optional.empty(); if (!candidateWorkersForReassignment.isEmpty()) { candidateWorkerLoad = pickCandidateWorkerForReassignment(completeWorkerAssignment); @@ -366,15 +371,15 @@ protected void handleLostAssignments(ConnectorsAndTasks lostAssignments, if (candidateWorkerLoad.isPresent()) { WorkerLoad workerLoad = candidateWorkerLoad.get(); + log.debug("A candidate worker has been found to assign lost tasks: {}", workerLoad.worker()); lostAssignments.connectors().forEach(workerLoad::assign); lostAssignments.tasks().forEach(workerLoad::assign); } else { + log.debug("No single candidate worker was found to assign lost tasks. Treating lost tasks as new tasks"); newSubmissions.connectors().addAll(lostAssignments.connectors()); newSubmissions.tasks().addAll(lostAssignments.tasks()); } - candidateWorkersForReassignment.clear(); - scheduledRebalance = 0; - delay = 0; + resetDelay(); } else { candidateWorkersForReassignment .addAll(candidateWorkersForReassignment(completeWorkerAssignment)); @@ -382,17 +387,29 @@ protected void handleLostAssignments(ConnectorsAndTasks lostAssignments, // a delayed rebalance is in progress, but it's not yet time to reassign // unaccounted resources delay = calculateDelay(now); + log.debug("Delayed rebalance in progress. Task reassignment is postponed. New computed rebalance delay: {}", delay); } else { // This means scheduledRebalance == 0 // We could also also extract the current minimum delay from the group, to make // independent of consecutive leader failures, but this optimization is skipped // at the moment delay = maxDelay; + log.debug("Resetting rebalance delay to the max: {}. scheduledRebalance: {} now: {} diff scheduledRebalance - now: {}", + delay, scheduledRebalance, now, scheduledRebalance - now); } scheduledRebalance = now + delay; } } + private void resetDelay() { + candidateWorkersForReassignment.clear(); + scheduledRebalance = 0; + if (delay != 0) { + log.debug("Resetting delay from previous value: {} to 0", delay); + } + delay = 0; + } + private Set candidateWorkersForReassignment(List completeWorkerAssignment) { return completeWorkerAssignment.stream() .filter(WorkerLoad::isEmpty) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java index a8ddbcc955b44..71f351df9ea63 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java @@ -56,7 +56,7 @@ public class WorkerCoordinator extends AbstractCoordinator implements Closeable private final Logger log; private final String restUrl; private final ConfigBackingStore configStorage; - private ExtendedAssignment assignmentSnapshot; + private volatile ExtendedAssignment assignmentSnapshot; private ClusterConfigState configSnapshot; private final WorkerRebalanceListener listener; private final ConnectProtocolCompatibility protocolCompatibility; @@ -66,6 +66,7 @@ public class WorkerCoordinator extends AbstractCoordinator implements Closeable private volatile ConnectProtocolCompatibility currentConnectProtocol; private final ConnectAssignor eagerAssignor; private final ConnectAssignor incrementalAssignor; + private final int coordinatorDiscoveryTimeoutMs; /** * Initialize the coordination manager. @@ -98,6 +99,7 @@ public WorkerCoordinator(GroupRebalanceConfig config, this.incrementalAssignor = new IncrementalCooperativeAssignor(logContext, time, maxDelay); this.eagerAssignor = new EagerAssignor(logContext); this.currentConnectProtocol = protocolCompatibility; + this.coordinatorDiscoveryTimeoutMs = config.heartbeatIntervalMs; } @Override @@ -124,7 +126,20 @@ public void poll(long timeout) { do { if (coordinatorUnknown()) { - ensureCoordinatorReady(time.timer(Long.MAX_VALUE)); + log.debug("Broker coordinator is marked unknown. Attempting discovery with a timeout of {}ms", + coordinatorDiscoveryTimeoutMs); + if (ensureCoordinatorReady(time.timer(coordinatorDiscoveryTimeoutMs))) { + log.debug("Broker coordinator is ready"); + } else { + log.debug("Can not connect to broker coordinator"); + final ExtendedAssignment localAssignmentSnapshot = assignmentSnapshot; + if (localAssignmentSnapshot != null && !localAssignmentSnapshot.failed()) { + log.info("Broker coordinator was unreachable for {}ms. Revoking previous assignment {} to " + + "avoid running tasks while not being a member the group", coordinatorDiscoveryTimeoutMs, localAssignmentSnapshot); + listener.onRevoked(localAssignmentSnapshot.leader(), localAssignmentSnapshot.connectors(), localAssignmentSnapshot.tasks()); + assignmentSnapshot = null; + } + } now = time.milliseconds(); } @@ -152,7 +167,8 @@ public void poll(long timeout) { @Override public JoinGroupRequestProtocolCollection metadata() { configSnapshot = configStorage.snapshot(); - ExtendedWorkerState workerState = new ExtendedWorkerState(restUrl, configSnapshot.offset(), assignmentSnapshot); + final ExtendedAssignment localAssignmentSnapshot = assignmentSnapshot; + ExtendedWorkerState workerState = new ExtendedWorkerState(restUrl, configSnapshot.offset(), localAssignmentSnapshot); switch (protocolCompatibility) { case EAGER: return ConnectProtocol.metadataRequest(workerState); @@ -180,17 +196,18 @@ protected void onJoinComplete(int generation, String memberId, String protocol, listener.onRevoked(newAssignment.leader(), newAssignment.revokedConnectors(), newAssignment.revokedTasks()); } - if (assignmentSnapshot != null) { - assignmentSnapshot.connectors().removeAll(newAssignment.revokedConnectors()); - assignmentSnapshot.tasks().removeAll(newAssignment.revokedTasks()); - log.debug("After revocations snapshot of assignment: {}", assignmentSnapshot); - newAssignment.connectors().addAll(assignmentSnapshot.connectors()); - newAssignment.tasks().addAll(assignmentSnapshot.tasks()); + final ExtendedAssignment localAssignmentSnapshot = assignmentSnapshot; + if (localAssignmentSnapshot != null) { + localAssignmentSnapshot.connectors().removeAll(newAssignment.revokedConnectors()); + localAssignmentSnapshot.tasks().removeAll(newAssignment.revokedTasks()); + log.debug("After revocations snapshot of assignment: {}", localAssignmentSnapshot); + newAssignment.connectors().addAll(localAssignmentSnapshot.connectors()); + newAssignment.tasks().addAll(localAssignmentSnapshot.tasks()); } log.debug("Augmented new assignment: {}", newAssignment); } assignmentSnapshot = newAssignment; - listener.onAssigned(assignmentSnapshot, generation); + listener.onAssigned(newAssignment, generation); } @Override @@ -204,19 +221,21 @@ protected Map performAssignment(String leaderId, String prot protected void onJoinPrepare(int generation, String memberId) { log.info("Rebalance started"); leaderState(null); + final ExtendedAssignment localAssignmentSnapshot = assignmentSnapshot; if (currentConnectProtocol == EAGER) { - log.debug("Revoking previous assignment {}", assignmentSnapshot); - if (assignmentSnapshot != null && !assignmentSnapshot.failed()) - listener.onRevoked(assignmentSnapshot.leader(), assignmentSnapshot.connectors(), assignmentSnapshot.tasks()); + log.debug("Revoking previous assignment {}", localAssignmentSnapshot); + if (localAssignmentSnapshot != null && !localAssignmentSnapshot.failed()) + listener.onRevoked(localAssignmentSnapshot.leader(), localAssignmentSnapshot.connectors(), localAssignmentSnapshot.tasks()); } else { log.debug("Cooperative rebalance triggered. Keeping assignment {} until it's " - + "explicitly revoked.", assignmentSnapshot); + + "explicitly revoked.", localAssignmentSnapshot); } } @Override protected boolean rejoinNeededOrPending() { - return super.rejoinNeededOrPending() || (assignmentSnapshot == null || assignmentSnapshot.failed()) || rejoinRequested; + final ExtendedAssignment localAssignmentSnapshot = assignmentSnapshot; + return super.rejoinNeededOrPending() || (localAssignmentSnapshot == null || localAssignmentSnapshot.failed()) || rejoinRequested; } @Override @@ -227,8 +246,19 @@ public String memberId() { return JoinGroupRequest.UNKNOWN_MEMBER_ID; } + /** + * Return the current generation. The generation refers to this worker's knowledge with + * respect to which generation is the latest one and, therefore, this information is local. + * + * @return the generation ID or -1 if no generation is defined + */ + public int generationId() { + return super.generation().generationId; + } + private boolean isLeader() { - return assignmentSnapshot != null && memberId().equals(assignmentSnapshot.leader()); + final ExtendedAssignment localAssignmentSnapshot = assignmentSnapshot; + return localAssignmentSnapshot != null && memberId().equals(localAssignmentSnapshot.leader()); } public String ownerUrl(String connector) { @@ -306,20 +336,22 @@ public WorkerCoordinatorMetrics(Metrics metrics, String metricGrpPrefix) { Measurable numConnectors = new Measurable() { @Override public double measure(MetricConfig config, long now) { - if (assignmentSnapshot == null) { + final ExtendedAssignment localAssignmentSnapshot = assignmentSnapshot; + if (localAssignmentSnapshot == null) { return 0.0; } - return assignmentSnapshot.connectors().size(); + return localAssignmentSnapshot.connectors().size(); } }; Measurable numTasks = new Measurable() { @Override public double measure(MetricConfig config, long now) { - if (assignmentSnapshot == null) { + final ExtendedAssignment localAssignmentSnapshot = assignmentSnapshot; + if (localAssignmentSnapshot == null) { return 0.0; } - return assignmentSnapshot.tasks().size(); + return localAssignmentSnapshot.tasks().size(); } }; diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java index 4819db5da7417..cc052c2907726 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java @@ -151,6 +151,10 @@ public void stop() { stop(false); } + /** + * Ensure that the connection to the broker coordinator is up and that the worker is an + * active member of the group. + */ public void ensureActive() { coordinator.poll(0); } 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 7cdfa7dcb9df0..4a704664b69db 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 @@ -17,6 +17,7 @@ package org.apache.kafka.connect.integration; import org.apache.kafka.connect.runtime.AbstractStatus; +import org.apache.kafka.connect.runtime.distributed.DistributedConfig; import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo; import org.apache.kafka.connect.storage.StringConverter; import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster; @@ -65,29 +66,27 @@ public class ConnectWorkerIntegrationTest { private static final int NUM_WORKERS = 3; private static final String CONNECTOR_NAME = "simple-source"; + private EmbeddedConnectCluster.Builder connectBuilder; private EmbeddedConnectCluster connect; + Map workerProps = new HashMap<>(); + Properties brokerProps = new Properties(); @Before public void setup() throws IOException { // setup Connect worker properties - Map workerProps = new HashMap<>(); workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS)); workerProps.put(CONNECTOR_CLIENT_POLICY_CLASS_CONFIG, "All"); // setup Kafka broker properties - Properties brokerProps = new Properties(); brokerProps.put("auto.create.topics.enable", String.valueOf(false)); // build a Connect cluster backed by Kafka and Zk - connect = new EmbeddedConnectCluster.Builder() + connectBuilder = new EmbeddedConnectCluster.Builder() .name("connect-cluster") .numWorkers(NUM_WORKERS) .workerProps(workerProps) .brokerProps(brokerProps) - .build(); - - // start the clusters - connect.start(); + .maskExitProcedures(true); // true is the default, setting here as example } @After @@ -102,6 +101,10 @@ public void close() { */ @Test public void testAddAndRemoveWorker() throws Exception { + connect = connectBuilder.build(); + // start the clusters + connect.start(); + int numTasks = 4; // create test topic connect.kafka().createTopic("test-topic", NUM_TOPIC_PARTITIONS); @@ -149,6 +152,10 @@ public void testAddAndRemoveWorker() throws Exception { */ @Test public void testRestartFailedTask() throws Exception { + connect = connectBuilder.build(); + // start the clusters + connect.start(); + int numTasks = 1; // Properties for the source connector. The task should fail at startup due to the bad broker address. @@ -180,6 +187,63 @@ public void testRestartFailedTask() throws Exception { CONNECTOR_SETUP_DURATION_MS, "Connector tasks are not all in running state."); } + /** + * Verify that a set of tasks restarts correctly after a broker goes offline and back online + */ + @Test + public void testBrokerCoordinator() throws Exception { + workerProps.put(DistributedConfig.SCHEDULED_REBALANCE_MAX_DELAY_MS_CONFIG, String.valueOf(5000)); + connect = connectBuilder.workerProps(workerProps).build(); + // start the clusters + connect.start(); + int numTasks = 4; + // create test topic + connect.kafka().createTopic("test-topic", NUM_TOPIC_PARTITIONS); + + // setup up props for the sink connector + Map props = new HashMap<>(); + props.put(CONNECTOR_CLASS_CONFIG, MonitorableSourceConnector.class.getSimpleName()); + props.put(TASKS_MAX_CONFIG, String.valueOf(numTasks)); + props.put("topic", "test-topic"); + props.put("throughput", String.valueOf(1)); + props.put("messages.per.poll", String.valueOf(10)); + props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName()); + props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName()); + + waitForCondition(() -> assertWorkersUp(NUM_WORKERS).orElse(false), + WORKER_SETUP_DURATION_MS, "Initial group of workers did not start in time."); + + // start a source connector + connect.configureConnector(CONNECTOR_NAME, props); + + waitForCondition(() -> assertConnectorAndTasksRunning(CONNECTOR_NAME, numTasks).orElse(false), + CONNECTOR_SETUP_DURATION_MS, "Connector tasks did not start in time."); + + connect.kafka().stopOnlyKafka(); + + waitForCondition(() -> assertWorkersUp(NUM_WORKERS).orElse(false), + WORKER_SETUP_DURATION_MS, "Group of workers did not remain the same after broker shutdown"); + + // Allow for the workers to discover that the coordinator is unavailable, wait is + // heartbeat timeout * 2 + 4sec + Thread.sleep(TimeUnit.SECONDS.toMillis(10)); + + connect.kafka().startOnlyKafkaOnSamePorts(); + + // Allow for the kafka brokers to come back online + Thread.sleep(TimeUnit.SECONDS.toMillis(10)); + + waitForCondition(() -> assertWorkersUp(NUM_WORKERS).orElse(false), + WORKER_SETUP_DURATION_MS, "Group of workers did not remain the same within the " + + "designated time."); + + // Allow for the workers to rebalance and reach a steady state + Thread.sleep(TimeUnit.SECONDS.toMillis(10)); + + waitForCondition(() -> assertConnectorAndTasksRunning(CONNECTOR_NAME, numTasks).orElse(false), + CONNECTOR_SETUP_DURATION_MS, "Connector tasks did not start in time."); + } + /** * Confirm that the requested number of workers is up and running. * diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java index c46d59bf8058a..a5e4ef0ca1bea 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java @@ -170,6 +170,8 @@ public void testTaskAssignmentWhenWorkerJoins() { verify(coordinator, times(rebalanceNum)).configSnapshot(); verify(coordinator, times(rebalanceNum)).leaderState(any()); + verify(coordinator, times(rebalanceNum)).generationId(); + verify(coordinator, times(rebalanceNum)).memberId(); } @Test @@ -234,6 +236,8 @@ public void testTaskAssignmentWhenWorkerLeavesPermanently() { verify(coordinator, times(rebalanceNum)).configSnapshot(); verify(coordinator, times(rebalanceNum)).leaderState(any()); + verify(coordinator, times(rebalanceNum)).generationId(); + verify(coordinator, times(rebalanceNum)).memberId(); } @Test @@ -314,6 +318,8 @@ public void testTaskAssignmentWhenWorkerBounces() { verify(coordinator, times(rebalanceNum)).configSnapshot(); verify(coordinator, times(rebalanceNum)).leaderState(any()); + verify(coordinator, times(rebalanceNum)).generationId(); + verify(coordinator, times(rebalanceNum)).memberId(); } @Test @@ -369,6 +375,8 @@ public void testTaskAssignmentWhenLeaderLeavesPermanently() { verify(coordinator, times(rebalanceNum)).configSnapshot(); verify(coordinator, times(rebalanceNum)).leaderState(any()); + verify(coordinator, times(rebalanceNum)).generationId(); + verify(coordinator, times(rebalanceNum)).memberId(); } @Test @@ -438,6 +446,8 @@ public void testTaskAssignmentWhenLeaderBounces() { verify(coordinator, times(rebalanceNum)).configSnapshot(); verify(coordinator, times(rebalanceNum)).leaderState(any()); + verify(coordinator, times(rebalanceNum)).generationId(); + verify(coordinator, times(rebalanceNum)).memberId(); } @Test @@ -507,6 +517,8 @@ public void testTaskAssignmentWhenFirstAssignmentAttemptFails() { verify(coordinator, times(rebalanceNum)).configSnapshot(); verify(coordinator, times(rebalanceNum)).leaderState(any()); + verify(coordinator, times(rebalanceNum)).generationId(); + verify(coordinator, times(rebalanceNum)).memberId(); } @Test @@ -563,6 +575,8 @@ public void testTaskAssignmentWhenSubsequentAssignmentAttemptFails() { verify(coordinator, times(rebalanceNum)).configSnapshot(); verify(coordinator, times(rebalanceNum)).leaderState(any()); + verify(coordinator, times(rebalanceNum)).generationId(); + verify(coordinator, times(rebalanceNum)).memberId(); } @Test @@ -596,6 +610,8 @@ public void testTaskAssignmentWhenConnectorsAreDeleted() { verify(coordinator, times(rebalanceNum)).configSnapshot(); verify(coordinator, times(rebalanceNum)).leaderState(any()); + verify(coordinator, times(rebalanceNum)).generationId(); + verify(coordinator, times(rebalanceNum)).memberId(); } @Test 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 948d54b57bd51..d36ae5b078e18 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 @@ -81,6 +81,8 @@ public class EmbeddedKafkaCluster extends ExternalResource { private final KafkaServer[] brokers; private final Properties brokerConfig; private final Time time = new MockTime(); + private final int[] currentBrokerPorts; + private final String[] currentBrokerLogDirs; private EmbeddedZookeeper zookeeper = null; private ListenerName listenerName = new ListenerName("PLAINTEXT"); @@ -89,6 +91,8 @@ public class EmbeddedKafkaCluster extends ExternalResource { public EmbeddedKafkaCluster(final int numBrokers, final Properties brokerConfig) { brokers = new KafkaServer[numBrokers]; + currentBrokerPorts = new int[numBrokers]; + currentBrokerLogDirs = new String[numBrokers]; this.brokerConfig = brokerConfig; } @@ -102,11 +106,20 @@ protected void after() { stop(); } + public void startOnlyKafkaOnSamePorts() throws IOException { + start(currentBrokerPorts, currentBrokerLogDirs); + } + private void start() throws IOException { + // pick a random port zookeeper = new EmbeddedZookeeper(); + Arrays.fill(currentBrokerPorts, 0); + Arrays.fill(currentBrokerLogDirs, null); + start(currentBrokerPorts, currentBrokerLogDirs); + } + private void start(int[] brokerPorts, String[] logDirs) throws IOException { brokerConfig.put(KafkaConfig$.MODULE$.ZkConnectProp(), zKConnectString()); - brokerConfig.put(KafkaConfig$.MODULE$.PortProp(), 0); // pick a random port putIfAbsent(brokerConfig, KafkaConfig$.MODULE$.HostNameProp(), "localhost"); putIfAbsent(brokerConfig, KafkaConfig$.MODULE$.DeleteTopicEnableProp(), true); @@ -121,8 +134,11 @@ private void start() throws IOException { for (int i = 0; i < brokers.length; i++) { brokerConfig.put(KafkaConfig$.MODULE$.BrokerIdProp(), i); - brokerConfig.put(KafkaConfig$.MODULE$.LogDirProp(), createLogDir()); + currentBrokerLogDirs[i] = logDirs[i] == null ? createLogDir() : currentBrokerLogDirs[i]; + brokerConfig.put(KafkaConfig$.MODULE$.LogDirProp(), currentBrokerLogDirs[i]); + brokerConfig.put(KafkaConfig$.MODULE$.PortProp(), brokerPorts[i]); brokers[i] = TestUtils.createServer(new KafkaConfig(brokerConfig, true), time); + currentBrokerPorts[i] = brokers[i].boundPort(listenerName); } Map producerProps = new HashMap<>(); @@ -132,8 +148,15 @@ private void start() throws IOException { producer = new KafkaProducer<>(producerProps); } + public void stopOnlyKafka() { + stop(false, false); + } + private void stop() { + stop(true, true); + } + private void stop(boolean deleteLogDirs, boolean stopZK) { try { producer.close(); } catch (Exception e) { @@ -151,19 +174,24 @@ private void stop() { } } - 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); + 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 { - zookeeper.shutdown(); + if (stopZK) { + zookeeper.shutdown(); + } } catch (Throwable t) { String msg = String.format("Could not shutdown zookeeper at %s", zKConnectString()); log.error(msg, t);