diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index a562eef2d0e93..18be1bb1421f9 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -92,7 +92,7 @@
-
+
diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java b/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
index ed4c0d98596cc..dc8f0f115bcda 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
@@ -23,6 +23,7 @@ public final class ClientRequest {
private final boolean expectResponse;
private final RequestSend request;
private final RequestCompletionHandler callback;
+ private final boolean isInitiatedByNetworkClient;
/**
* @param createdMs The unix timestamp in milliseconds for the time at which this request was created.
@@ -30,17 +31,35 @@ public final class ClientRequest {
* @param request The request
* @param callback A callback to execute when the response has been received (or null if no callback is necessary)
*/
- public ClientRequest(long createdMs, boolean expectResponse, RequestSend request, RequestCompletionHandler callback) {
+ public ClientRequest(long createdMs, boolean expectResponse, RequestSend request,
+ RequestCompletionHandler callback) {
+ this(createdMs, expectResponse, request, callback, false);
+ }
+
+ /**
+ * @param createdMs The unix timestamp in milliseconds for the time at which this request was created.
+ * @param expectResponse Should we expect a response message or is this request complete once it is sent?
+ * @param request The request
+ * @param callback A callback to execute when the response has been received (or null if no callback is necessary)
+ * @param isInitiatedByNetworkClient Is request initiated by network client, if yes, its
+ * response will be consumed by network client
+ */
+ public ClientRequest(long createdMs, boolean expectResponse, RequestSend request,
+ RequestCompletionHandler callback, boolean isInitiatedByNetworkClient) {
this.createdMs = createdMs;
this.callback = callback;
this.request = request;
this.expectResponse = expectResponse;
+ this.isInitiatedByNetworkClient = isInitiatedByNetworkClient;
}
@Override
public String toString() {
- return "ClientRequest(expectResponse=" + expectResponse + ", callback=" + callback + ", request=" + request
- + ")";
+ return "ClientRequest(expectResponse=" + expectResponse +
+ ", callback=" + callback +
+ ", request=" + request +
+ (isInitiatedByNetworkClient ? ", isInitiatedByNetworkClient" : "") +
+ ")";
}
public boolean expectResponse() {
@@ -63,4 +82,8 @@ public long createdTime() {
return createdMs;
}
+ public boolean isInitiatedByNetworkClient() {
+ return isInitiatedByNetworkClient;
+ }
+
}
\ No newline at end of file
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
index 48fe7961e2215..0e51d7bd461d2 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -378,7 +378,7 @@ private void handleCompletedReceives(List responses, long now) {
short apiKey = req.request().header().apiKey();
Struct body = (Struct) ProtoUtils.currentResponseSchema(apiKey).read(receive.payload());
correlate(req.request().header(), header);
- if (apiKey == ApiKeys.METADATA.id) {
+ if (apiKey == ApiKeys.METADATA.id && req.isInitiatedByNetworkClient()) {
handleMetadataResponse(req.request().header(), body, now);
} else {
// need to add body/header to response here
@@ -454,7 +454,7 @@ private void correlate(RequestHeader requestHeader, ResponseHeader responseHeade
private ClientRequest metadataRequest(long now, String node, Set topics) {
MetadataRequest metadata = new MetadataRequest(new ArrayList(topics));
RequestSend send = new RequestSend(node, nextRequestHeader(ApiKeys.METADATA), metadata.toStruct());
- return new ClientRequest(now, true, send, null);
+ return new ClientRequest(now, true, send, null, true);
}
/**
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java
index 252b759c0801f..23e410b7d93f1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java
@@ -113,6 +113,11 @@ public interface Consumer extends Closeable {
*/
public List partitionsFor(String topic);
+ /**
+ * @see KafkaConsumer#listTopics()
+ */
+ public Map> listTopics();
+
/**
* @see KafkaConsumer#close()
*/
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index bea3d737c51be..923ff999d1b04 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -1024,6 +1024,22 @@ public List partitionsFor(String topic) {
}
}
+ /**
+ * Get metadata about partitions for all topics. This method will issue a remote call to the
+ * server.
+ *
+ * @return The map of topics and its partitions
+ */
+ @Override
+ public Map> listTopics() {
+ acquire();
+ try {
+ return fetcher.getAllTopics(requestTimeoutMs);
+ } finally {
+ release();
+ }
+ }
+
@Override
public void close() {
acquire();
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
index c14eed1e95f2e..5b22fa0bcb479 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
@@ -177,6 +177,12 @@ public synchronized List partitionsFor(String topic) {
return parts;
}
+ @Override
+ public Map> listTopics() {
+ ensureNotClosed();
+ return partitions;
+ }
+
public synchronized void updatePartitions(String topic, List partitions) {
ensureNotClosed();
this.partitions.put(topic, partitions);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
index d2a0e2be67867..9f71451e1b075 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
@@ -39,6 +39,8 @@
import org.apache.kafka.common.requests.FetchResponse;
import org.apache.kafka.common.requests.ListOffsetRequest;
import org.apache.kafka.common.requests.ListOffsetResponse;
+import org.apache.kafka.common.requests.MetadataRequest;
+import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
@@ -160,6 +162,48 @@ public void updateFetchPositions(Set partitions) {
}
}
+
+
+ /**
+ * Get metadata for all topics present in Kafka cluster
+ *
+ * @param timeout time for which getting all topics is attempted
+ * @return The map of topics and its partitions
+ */
+ public Map> getAllTopics(long timeout) {
+ final HashMap> topicsPartitionInfos = new HashMap<>();
+ long startTime = time.milliseconds();
+
+ while (time.milliseconds() - startTime < timeout) {
+ final Node node = client.leastLoadedNode();
+ if (node != null) {
+ MetadataRequest metadataRequest = new MetadataRequest(Collections.emptyList());
+ final RequestFuture requestFuture =
+ client.send(node, ApiKeys.METADATA, metadataRequest);
+
+ client.poll(requestFuture);
+
+ if (requestFuture.succeeded()) {
+ MetadataResponse response =
+ new MetadataResponse(requestFuture.value().responseBody());
+
+ for (String topic : response.cluster().topics())
+ topicsPartitionInfos.put(
+ topic, response.cluster().availablePartitionsForTopic(topic));
+
+ return topicsPartitionInfos;
+ }
+
+ if (!requestFuture.isRetriable())
+ throw requestFuture.exception();
+ }
+
+ Utils.sleep(retryBackoffMs);
+ }
+
+ return topicsPartitionInfos;
+ }
+
/**
* Reset offsets for the given partition using the offset reset strategy.
*
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
index 4d9a425201115..8a2cb1237eada 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
@@ -15,6 +15,7 @@
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.TopicPartition;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -83,7 +84,8 @@ public void unsubscribe(String topic) {
throw new IllegalStateException("Topic " + topic + " was never subscribed to.");
this.subscribedTopics.remove(topic);
this.needsPartitionAssignment = true;
- for (TopicPartition tp: assignedPartitions())
+ final List existingAssignedPartitions = new ArrayList<>(assignedPartitions());
+ for (TopicPartition tp: existingAssignedPartitions)
if (topic.equals(tp.topic()))
clearPartition(tp);
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java b/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java
index 249d6b8a65745..5fe882151613e 100644
--- a/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java
@@ -12,7 +12,7 @@
*/
package org.apache.kafka.clients;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.errors.TimeoutException;
@@ -27,11 +27,11 @@ public class MetadataTest {
private long refreshBackoffMs = 100;
private long metadataExpireMs = 1000;
private Metadata metadata = new Metadata(refreshBackoffMs, metadataExpireMs);
- private AtomicBoolean backgroundError = new AtomicBoolean(false);
+ private AtomicReference backgroundError = new AtomicReference();
@After
public void tearDown() {
- assertFalse(backgroundError.get());
+ assertNull("Exception in background thread : " + backgroundError.get(), backgroundError.get());
}
@Test
@@ -48,7 +48,15 @@ public void testMetadata() throws Exception {
Thread t2 = asyncFetch(topic);
assertTrue("Awaiting update", t1.isAlive());
assertTrue("Awaiting update", t2.isAlive());
- metadata.update(TestUtils.singletonCluster(topic, 1), time);
+ // Perform metadata update when an update is requested on the async fetch thread
+ // This simulates the metadata update sequence in KafkaProducer
+ while (t1.isAlive() || t2.isAlive()) {
+ if (metadata.timeToNextUpdate(time) == 0) {
+ metadata.update(TestUtils.singletonCluster(topic, 1), time);
+ time += refreshBackoffMs;
+ }
+ Thread.sleep(1);
+ }
t1.join();
t2.join();
assertFalse("No update needed.", metadata.timeToNextUpdate(time) == 0);
@@ -106,7 +114,7 @@ public void run() {
try {
metadata.awaitUpdate(metadata.requestUpdate(), refreshBackoffMs);
} catch (Exception e) {
- backgroundError.set(true);
+ backgroundError.set(e.toString());
}
}
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
index 4002679cbc8cc..06e2990636522 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
@@ -22,6 +22,7 @@
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.protocol.Errors;
@@ -29,6 +30,7 @@
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.requests.FetchResponse;
+import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.test.TestUtils;
@@ -180,6 +182,17 @@ public void testFetchOutOfRange() {
assertEquals(null, subscriptions.consumed(tp));
}
+ @Test
+ public void testGetAllTopics() throws InterruptedException {
+ // sending response before request, as getAllTopics is a blocking call
+ client.prepareResponse(
+ new MetadataResponse(cluster, Collections.emptyMap()).toStruct());
+
+ Map> allTopics = fetcher.getAllTopics(5000L);
+
+ assertEquals(cluster.topics().size(), allTopics.size());
+ }
+
private Struct fetchResponse(ByteBuffer buffer, short error, long hw) {
FetchResponse response = new FetchResponse(Collections.singletonMap(tp, new FetchResponse.PartitionData(error, hw, buffer)));
return response.toStruct();
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java
index 319751c374ccd..c47f3fb699d7a 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java
@@ -73,6 +73,27 @@ public void topicSubscription() {
assertAllPositions(tp0, null);
assertEquals(Collections.singleton(tp1), state.assignedPartitions());
}
+
+ @Test
+ public void topicUnsubscription() {
+ final String topic = "test";
+ state.subscribe(topic);
+ assertEquals(1, state.subscribedTopics().size());
+ assertTrue(state.assignedPartitions().isEmpty());
+ assertTrue(state.partitionsAutoAssigned());
+ state.changePartitionAssignment(asList(tp0));
+ state.committed(tp0, 1);
+ state.fetched(tp0, 1);
+ state.consumed(tp0, 1);
+ assertAllPositions(tp0, 1L);
+ state.changePartitionAssignment(asList(tp1));
+ assertAllPositions(tp0, null);
+ assertEquals(Collections.singleton(tp1), state.assignedPartitions());
+
+ state.unsubscribe(topic);
+ assertEquals(0, state.subscribedTopics().size());
+ assertTrue(state.assignedPartitions().isEmpty());
+ }
@Test(expected = IllegalArgumentException.class)
public void cantChangeFetchPositionForNonAssignedPartition() {
diff --git a/config/server.properties b/config/server.properties
index 80ee2fc6e94a1..9e0a43de4fce0 100644
--- a/config/server.properties
+++ b/config/server.properties
@@ -121,3 +121,5 @@ zookeeper.connect=localhost:2181
# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000
+
+auto.create.topics.enable=true
diff --git a/core/src/main/scala/kafka/javaapi/consumer/ConsumerConnector.java b/core/src/main/scala/kafka/javaapi/consumer/ConsumerConnector.java
index ca74ca8abf034..444cd1d4b34e6 100644
--- a/core/src/main/scala/kafka/javaapi/consumer/ConsumerConnector.java
+++ b/core/src/main/scala/kafka/javaapi/consumer/ConsumerConnector.java
@@ -75,6 +75,12 @@ public interface ConsumerConnector {
*/
public void commitOffsets(Map offsetsToCommit, boolean retryOnFailure);
+ /**
+ * Wire in a consumer rebalance listener to be executed when consumer rebalance occurs.
+ * @param listener The consumer rebalance listener to wire in
+ */
+ public void setConsumerRebalanceListener(ConsumerRebalanceListener listener);
+
/**
* Shut down the connector
*/
diff --git a/core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala b/core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala
index 3d52f62c88a50..c39fbfe483b82 100644
--- a/core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala
+++ b/core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala
@@ -107,6 +107,8 @@ object ConsumerOffsetChecker extends Logging {
}
def main(args: Array[String]) {
+ warn("WARNING: ConsumerOffsetChecker is deprecated and will be dropped in releases following 0.9.0. Use ConsumerGroupCommand instead.")
+
val parser = new OptionParser()
val zkConnectOpt = parser.accepts("zookeeper", "ZooKeeper connect string.").
diff --git a/core/src/test/scala/integration/kafka/api/ConsumerTest.scala b/core/src/test/scala/integration/kafka/api/ConsumerTest.scala
index 3eb5f95731a3f..0c2755f724098 100644
--- a/core/src/test/scala/integration/kafka/api/ConsumerTest.scala
+++ b/core/src/test/scala/integration/kafka/api/ConsumerTest.scala
@@ -186,6 +186,24 @@ class ConsumerTest extends IntegrationTestHarness with Logging {
assertNull(this.consumers(0).partitionsFor("non-exist-topic"))
}
+ def testListTopics() {
+ val numParts = 2
+ val topic1: String = "part-test-topic-1"
+ val topic2: String = "part-test-topic-2"
+ val topic3: String = "part-test-topic-3"
+ TestUtils.createTopic(this.zkClient, topic1, numParts, 1, this.servers)
+ TestUtils.createTopic(this.zkClient, topic2, numParts, 1, this.servers)
+ TestUtils.createTopic(this.zkClient, topic3, numParts, 1, this.servers)
+
+ val topics = this.consumers.head.listTopics()
+ assertNotNull(topics)
+ assertEquals(5, topics.size())
+ assertEquals(5, topics.keySet().size())
+ assertEquals(2, topics.get(topic1).length)
+ assertEquals(2, topics.get(topic2).length)
+ assertEquals(2, topics.get(topic3).length)
+ }
+
def testPartitionReassignmentCallback() {
val callback = new TestConsumerReassignmentCallback()
this.consumerConfig.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "100"); // timeout quickly to avoid slow test
@@ -217,6 +235,25 @@ class ConsumerTest extends IntegrationTestHarness with Logging {
consumer0.close()
}
+ def testUnsubscribeTopic() {
+ val callback = new TestConsumerReassignmentCallback()
+ this.consumerConfig.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "100"); // timeout quickly to avoid slow test
+ val consumer0 = new KafkaConsumer(this.consumerConfig, callback, new ByteArrayDeserializer(), new ByteArrayDeserializer())
+
+ try {
+ consumer0.subscribe(topic)
+
+ // the initial subscription should cause a callback execution
+ while (callback.callsToAssigned == 0)
+ consumer0.poll(50)
+
+ consumer0.unsubscribe(topic)
+ assertEquals(0, consumer0.subscriptions.size())
+ } finally {
+ consumer0.close()
+ }
+ }
+
private class TestConsumerReassignmentCallback extends ConsumerRebalanceCallback {
var callsToAssigned = 0
var callsToRevoked = 0
diff --git a/tests/kafkatest/process_signal.py b/tests/kafkatest/process_signal.py
new file mode 100644
index 0000000000000..b0825c9bcbd67
--- /dev/null
+++ b/tests/kafkatest/process_signal.py
@@ -0,0 +1,28 @@
+# Copyright 2015 Confluent Inc.
+#
+# Licensed 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.
+
+
+"""Commonly used process control signals.
+
+Note that using the python signal module has lead to errors because
+it maps signals to integers, and this mapping can be different on
+different machines.
+
+Using the signal name seems to be more robust across different Unix/Linux
+base operating systems.
+"""
+SIGTERM = "SIGTERM"
+SIGKILL = "SIGKILL"
+SIGSTOP = "SIGSTOP"
+SIGCONT = "SIGCONT"
\ No newline at end of file
diff --git a/tests/kafkatest/sanity_checks/README.md b/tests/kafkatest/sanity_checks/README.md
new file mode 100644
index 0000000000000..73006d23aeb6c
--- /dev/null
+++ b/tests/kafkatest/sanity_checks/README.md
@@ -0,0 +1,6 @@
+This package contains small tests to sanity check the functionality of services like KafkaService.
+
+When writing system or integration tests, it is helpful to know that the Service subclasses on which you rely
+work the way you think they work. Therefore, it is useful to have a class of tests which only exercise
+small facets of your Service subclasses. Such tests go in this directory, and can be run using ducktape just
+like the system tests.
\ No newline at end of file
diff --git a/tests/kafkatest/sanity_checks/__init__.py b/tests/kafkatest/sanity_checks/__init__.py
new file mode 100644
index 0000000000000..1896e9e3bf471
--- /dev/null
+++ b/tests/kafkatest/sanity_checks/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2015 Confluent Inc.
+#
+# Licensed 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.
diff --git a/tests/kafkatest/sanity_checks/kafka_service_test.py b/tests/kafkatest/sanity_checks/kafka_service_test.py
new file mode 100644
index 0000000000000..72c1065890739
--- /dev/null
+++ b/tests/kafkatest/sanity_checks/kafka_service_test.py
@@ -0,0 +1,92 @@
+# Copyright 2015 Confluent Inc.
+#
+# Licensed 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.
+
+
+from ducktape.tests.test import Test
+from ducktape.utils.util import wait_until
+from ducktape.mark import parametrize
+
+
+from kafkatest.services.kafka import KafkaService
+from kafkatest.services.zookeeper import ZookeeperService
+from kafkatest.process_signal import SIGTERM, SIGKILL, SIGSTOP
+
+
+class KafkaServiceTest(Test):
+ def __init__(self, test_context):
+ super(KafkaServiceTest, self).__init__(test_context=test_context)
+
+ self.zk = ZookeeperService(self.test_context, num_nodes=1)
+ self.kafka = KafkaService(self.test_context, num_nodes=1, zk=self.zk)
+
+ @parametrize(num_nodes=1)
+ @parametrize(num_nodes=3)
+ def test_start(self, num_nodes=1):
+ """Check that simply starting the service works"""
+
+ self.zk.start()
+ self.kafka.num_nodes = num_nodes
+ self.kafka.start()
+
+ assert self.kafka.all_alive()
+ for node in self.kafka.nodes:
+ assert len(self.kafka.pids(node)) == 1
+
+ def test_stop_node(self):
+ """Check that stopping a node works, even with a SIGSTOPed process."""
+ self.zk.start()
+ self.kafka.start()
+
+ node = self.kafka.nodes[0]
+ self.kafka.signal_node(node, sig=SIGSTOP)
+ self.kafka.stop_node(self.kafka.nodes[0])
+
+ if not wait_until(lambda: self.kafka.dead(node), timeout_sec=5, backoff_sec=.5):
+ raise RuntimeError("Kafka node failed to stop in a reasonable amount of time.")
+
+ @parametrize(sig=SIGTERM)
+ @parametrize(sig=SIGKILL)
+ def test_signal_node(self, sig=SIGTERM):
+ self.zk.start()
+ self.kafka.start()
+ assert self.kafka.all_alive()
+
+ node = self.kafka.nodes[0]
+ self.kafka.signal_node(node, sig=sig)
+ if not wait_until(lambda: self.kafka.dead(node), timeout_sec=5, backoff_sec=.5):
+ raise RuntimeError("Kafka node failed to stop in a reasonable amount of time.")
+
+ @parametrize(sig=SIGTERM)
+ # @parametrize(sig=SIGKILL)
+ def test_bounce_node(self, sig=SIGTERM):
+ self.zk.start()
+ self.kafka.num_nodes = 3
+ self.kafka.start()
+
+ node = self.kafka.nodes[0]
+ old_pid = self.kafka.pids(node)
+
+ self.kafka.bounce_node(
+ node, sig=sig, condition=lambda: self.kafka.dead(node), condition_timeout_sec=10, condition_backoff_sec=.5)
+ assert self.kafka.alive(node)
+
+ new_pid = self.kafka.pids(node)[0]
+ assert new_pid != old_pid
+
+ def test_find_controller(self):
+ self.zk.start()
+ self.kafka.start()
+
+ assert self.kafka.controller() == self.kafka.nodes[0]
+
diff --git a/tests/kafkatest/services/console_consumer.py b/tests/kafkatest/services/console_consumer.py
index 33ef4eaeb2e81..01a502967afcb 100644
--- a/tests/kafkatest/services/console_consumer.py
+++ b/tests/kafkatest/services/console_consumer.py
@@ -91,6 +91,7 @@ def __init__(self, context, num_nodes, kafka, topic, message_validator=is_int, f
"""
super(ConsoleConsumer, self).__init__(context, num_nodes)
self.kafka = kafka
+ self.topic = topic
self.args = {
'topic': topic,
}
@@ -132,7 +133,6 @@ def _worker(self, idx, node):
msg = line.strip()
msg = self.message_validator(msg)
if msg is not None:
- self.logger.debug("consumed a message: " + str(msg))
self.messages_consumed[idx].append(msg)
def start_node(self, node):
diff --git a/tests/kafkatest/services/kafka.py b/tests/kafkatest/services/kafka.py
index 34ec5ef95de9a..33736b2d1b2a4 100644
--- a/tests/kafkatest/services/kafka.py
+++ b/tests/kafkatest/services/kafka.py
@@ -14,10 +14,13 @@
# limitations under the License.
from ducktape.services.service import Service
+from ducktape.utils.util import wait_until
+from kafkatest.process_signal import *
+
+from concurrent import futures
import json
import re
-import signal
import time
@@ -27,8 +30,11 @@ class KafkaService(Service):
"kafka_log": {
"path": "/mnt/kafka.log",
"collect_default": True},
+ "kafka_operational_logs": {
+ "path": "/mnt/kafka-operational-logs",
+ "collect_default": True},
"kafka_data": {
- "path": "/mnt/kafka-logs",
+ "path": "/mnt/kafka-data-logs",
"collect_default": False}
}
@@ -45,8 +51,12 @@ def __init__(self, context, num_nodes, zk, topics=None):
def start(self):
super(KafkaService, self).start()
+ if not wait_until(lambda: self.all_alive(), timeout_sec=20, backoff_sec=.5):
+ raise RuntimeError("Timed out waiting for Kafka cluster to start.")
+
# Create topics if necessary
if self.topics is not None:
+
for topic, topic_cfg in self.topics.items():
if topic_cfg is None:
topic_cfg = {}
@@ -54,49 +64,83 @@ def start(self):
topic_cfg["topic"] = topic
self.create_topic(topic_cfg)
+ def stop_node(self, node):
+ pids = self.pids(node)
+
+ for pid in pids:
+ node.account.signal(pid, SIGTERM, allow_fail=False)
+
+ if wait_until(lambda: self.dead(node), timeout_sec=5, backoff_sec=.5):
+ return
+
+ # SIGTERM didn't succeed - try the more aggressive SIGKILL
+ for pid in pids:
+ node.account.signal(pid, SIGKILL, allow_fail=False)
+
+ if not wait_until(lambda: self.dead(node), timeout_sec=5, backoff_sec=.5):
+ raise RuntimeError("Failed to kill Kafka process on " + str(node.account))
+
+ def clean_node(self, node):
+ node.account.ssh(
+ "rm -rf /mnt/*",
+ allow_fail=False)
+
+ def all_alive(self):
+ """Are all nodes in this cluster alive and communicating?"""
+ for node in self.nodes:
+ if not self.alive(node):
+ return False
+ return True
+
+ def alive(self, node):
+ """Is the kafka process on this node awake and ready to communicate?"""
+ try:
+ # Try opening tcp connection and immediately closing by sending EOF
+ node.account.ssh("echo EOF | nc %s %d" % (node.account.hostname, 9092), allow_fail=False)
+ return True
+ except:
+ return False
+
+ def dead(self, node):
+ """Test of 'deadness' of a node. This is often not the same as 'not alive'."""
+ return len(self.pids(node)) == 0
+
def start_node(self, node):
+ assert self.dead(node), "Called start_node on a node which has a Kafka process that is not dead: " + \
+ str(node.account)
+
props_file = self.render('kafka.properties', node=node, broker_id=self.idx(node))
- self.logger.info("kafka.properties:")
- self.logger.info(props_file)
+ self.logger.debug("kafka.properties:")
+ self.logger.debug(props_file)
node.account.create_file("/mnt/kafka.properties", props_file)
- cmd = "/opt/kafka/bin/kafka-server-start.sh /mnt/kafka.properties 1>> /mnt/kafka.log 2>> /mnt/kafka.log & echo $! > /mnt/kafka.pid"
- self.logger.debug("Attempting to start KafkaService on %s with command: %s" % (str(node.account), cmd))
- node.account.ssh(cmd)
- time.sleep(5)
- if len(self.pids(node)) == 0:
- raise Exception("No process ids recorded on node %s" % str(node))
+ cmd = "export LOG_DIR=/mnt/kafka-operational-logs/; "
+ cmd += "/opt/kafka/bin/kafka-server-start.sh /mnt/kafka.properties 1>> /mnt/kafka.log 2>> /mnt/kafka.log &"
+ self.logger.debug("Attempting to start KafkaService on %s" % str(node.account))
+
+ with futures.ThreadPoolExecutor(max_workers=1) as executor:
+ future = executor.submit(lambda: node.account.ssh(cmd))
def pids(self, node):
"""Return process ids associated with running processes on the given node."""
try:
- return [pid for pid in node.account.ssh_capture("cat /mnt/kafka.pid", callback=int)]
+ cmd = "ps ax | grep java | grep -i 'kafka\.properties' | grep -v grep | awk '{print $1}'"
+ pids = [pid for pid in node.account.ssh_capture(cmd, allow_fail=True, callback=lambda x: int(x.strip()))]
+ return pids
except:
return []
- def signal_node(self, node, sig=signal.SIGTERM):
+ def signal_node(self, node, sig=SIGTERM):
pids = self.pids(node)
for pid in pids:
node.account.signal(pid, sig)
- def signal_leader(self, topic, partition=0, sig=signal.SIGTERM):
+ def signal_leader(self, topic, partition=0, sig=SIGTERM):
leader = self.leader(topic, partition)
self.signal_node(leader, sig)
- def stop_node(self, node, clean_shutdown=True):
- pids = self.pids(node)
- sig = signal.SIGTERM if clean_shutdown else signal.SIGKILL
-
- for pid in pids:
- node.account.signal(pid, sig, allow_fail=False)
-
- node.account.ssh("rm -f /mnt/kafka.pid", allow_fail=False)
-
- def clean_node(self, node):
- node.account.ssh("rm -rf /mnt/kafka-logs /mnt/kafka.properties /mnt/kafka.log /mnt/kafka.pid", allow_fail=False)
-
def create_topic(self, topic_cfg):
- node = self.nodes[0] # any node is fine here
+ node = self.nodes[0] # any node is fine here
self.logger.info("Creating topic %s with settings %s", topic_cfg["topic"], topic_cfg)
cmd = "/opt/kafka/bin/kafka-topics.sh --zookeeper %(zk_connect)s --create "\
@@ -111,11 +155,12 @@ def create_topic(self, topic_cfg):
for config_name, config_value in topic_cfg["configs"].items():
cmd += " --config %s=%s" % (config_name, str(config_value))
- self.logger.info("Running topic creation command...\n%s" % cmd)
+ self.logger.info("Running topic creation command...\n")
node.account.ssh(cmd)
time.sleep(1)
self.logger.info("Checking to see if topic was properly created...\n%s" % cmd)
+
for line in self.describe_topic(topic_cfg["topic"]).split("\n"):
self.logger.info(line)
@@ -129,8 +174,7 @@ def describe_topic(self, topic):
return output
def verify_reassign_partitions(self, reassignment):
- """Run the reassign partitions admin tool in "verify" mode
- """
+ """Run the reassign partitions admin tool in "verify" mode."""
node = self.nodes[0]
json_file = "/tmp/" + str(time.time()) + "_reassign.json"
@@ -158,12 +202,10 @@ def verify_reassign_partitions(self, reassignment):
if re.match(".*is in progress.*", output) is not None:
return False
-
return True
def execute_reassign_partitions(self, reassignment):
- """Run the reassign partitions admin tool in "verify" mode
- """
+ """Run the reassign partitions admin tool in "verify" mode."""
node = self.nodes[0]
json_file = "/tmp/" + str(time.time()) + "_reassign.json"
@@ -181,7 +223,7 @@ def execute_reassign_partitions(self, reassignment):
cmd += " && sleep 1 && rm -f %s" % json_file
# send command
- self.logger.info("Executing parition reassignment...")
+ self.logger.info("Executing partition reassignment...")
self.logger.debug(cmd)
output = ""
for line in node.account.ssh_capture(cmd):
@@ -190,38 +232,67 @@ def execute_reassign_partitions(self, reassignment):
self.logger.debug("Verify partition reassignment:")
self.logger.debug(output)
- def restart_node(self, node, wait_sec=0, clean_shutdown=True):
- """Restart the given node, waiting wait_sec in between stopping and starting up again."""
- self.stop_node(node, clean_shutdown)
- time.sleep(wait_sec)
- self.start_node(node)
+ def bounce_node(self, node, sig=SIGTERM, condition=None, condition_timeout_sec=5, condition_backoff_sec=.5):
+ """Helper method for the very common test task of bouncing a kafka node.
+
+ :param node A node in the service
+ :param sig Process control signal. Preferably of the form "SIGSTOP" rather than 17, since the integer corresponding
+ to a given signal is os-dependant.
+ :param condition Wait for this condition to be true before restarting
+ :param condition_timeout_sec Max time to wait for wake condition to become true
+ :param condition_backoff_sec
+ """
+ assert sig in {SIGKILL, SIGSTOP, SIGTERM}
+ self.signal_node(node, sig)
+
+ if condition:
+ # Wait until some condition is true before bringing the node back up
+ if not wait_until(condition, timeout_sec=condition_timeout_sec, backoff_sec=condition_backoff_sec):
+ raise RuntimeError("Timed out waiting for " + str(condition))
+
+ if sig == SIGSTOP:
+ self.signal_node(node, "SIGCONT")
+ else:
+ self.start_node(node)
+
+ if not wait_until(lambda: self.alive(node), timeout_sec=condition_timeout_sec, backoff_sec=condition_backoff_sec):
+ raise RuntimeError("Timed out waiting for %s to restart." % str(node.account))
def leader(self, topic, partition=0):
""" Get the leader replica for the given topic and partition.
+
+ Return None if there is currently no leader.
"""
- cmd = "/opt/kafka/bin/kafka-run-class.sh kafka.tools.ZooKeeperMainWrapper -server %s " \
- % self.zk.connect_setting()
- cmd += "get /brokers/topics/%s/partitions/%d/state" % (topic, partition)
- self.logger.debug(cmd)
+ topic_partition_data = self.zk.get_data("/brokers/topics/%s/partitions/%d/state" % (topic, partition))
+ if topic_partition_data is None:
+ return None
- node = self.nodes[0]
- self.logger.debug("Querying zookeeper to find leader replica for topic %s: \n%s" % (cmd, topic))
- partition_state = None
- for line in node.account.ssh_capture(cmd):
- match = re.match("^({.+})$", line)
- if match is not None:
- partition_state = match.groups()[0]
- break
+ leader_idx = int(topic_partition_data["leader"])
+ if leader_idx < 0:
+ return None
+
+ return self.get_node(leader_idx)
+
+ def controller(self):
+ """
+ Get the current controller
- if partition_state is None:
- raise Exception("Error finding partition state for topic %s and partition %d." % (topic, partition))
+ This may return None if the "/controller" path is empty
+ """
+ controller_data = self.zk.get_data("/controller")
+ if controller_data is None:
+ # raise Exception("Error finding controller.")
+ return None
- partition_state = json.loads(partition_state)
- self.logger.info(partition_state)
+ controller_idx = int(controller_data["brokerid"])
+ if controller_idx < 0:
+ return None
- leader_idx = int(partition_state["leader"])
- self.logger.info("Leader for topic %s and partition %d is now: %d" % (topic, partition, leader_idx))
- return self.get_node(leader_idx)
+ controller_node = self.get_node(controller_idx)
+
+ if controller_node is None:
+ raise Exception("Found no node corresponding to the id: %d" % controller_idx)
+ return controller_node
def bootstrap_servers(self):
- return ','.join([node.account.hostname + ":9092" for node in self.nodes])
\ No newline at end of file
+ return ','.join([node.account.hostname + ":9092" for node in self.nodes])
diff --git a/tests/kafkatest/services/templates/kafka.properties b/tests/kafkatest/services/templates/kafka.properties
index db1077a4a4eb8..c49fd8b4e1d0b 100644
--- a/tests/kafkatest/services/templates/kafka.properties
+++ b/tests/kafkatest/services/templates/kafka.properties
@@ -14,108 +14,27 @@
# limitations under the License.
# see kafka.server.KafkaConfig for additional details and defaults
-############################# Server Basics #############################
-# The id of the broker. This must be set to a unique integer for each broker.
broker.id={{ broker_id }}
-
-############################# Socket Server Settings #############################
-
-# The port the socket server listens on
port=9092
-
-# Hostname the broker will bind to. If not set, the server will bind to all interfaces
#host.name=localhost
-
-# Hostname the broker will advertise to producers and consumers. If not set, it uses the
-# value for "host.name" if configured. Otherwise, it will use the value returned from
-# java.net.InetAddress.getCanonicalHostName().
advertised.host.name={{ node.account.hostname }}
-
-# The port to publish to ZooKeeper for clients to use. If this is not set,
-# it will publish the same port that the broker binds to.
#advertised.port=
-
-# The number of threads handling network requests
num.network.threads=3
-
-# The number of threads doing disk I/O
num.io.threads=8
-
-# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400
-
-# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=65536
-
-# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600
-
-
-############################# Log Basics #############################
-
-# A comma seperated list of directories under which to store log files
log.dirs=/mnt/kafka-logs
-
-# The default number of log partitions per topic. More partitions allow greater
-# parallelism for consumption, but this will also result in more files across
-# the brokers.
num.partitions=1
-
-# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
-# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1
-
-############################# Log Flush Policy #############################
-
-# Messages are immediately written to the filesystem but by default we only fsync() to sync
-# the OS cache lazily. The following configurations control the flush of data to disk.
-# There are a few important trade-offs here:
-# 1. Durability: Unflushed data may be lost if you are not using replication.
-# 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
-# 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
-# The settings below allow one to configure the flush policy to flush data after a period of time or
-# every N messages (or both). This can be done globally and overridden on a per-topic basis.
-
-# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000
-
-# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000
-
-############################# Log Retention Policy #############################
-
-# The following configurations control the disposal of log segments. The policy can
-# be set to delete segments after a period of time, or after a given size has accumulated.
-# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
-# from the end of the log.
-
-# The minimum age of a log file to be eligible for deletion
log.retention.hours=168
-
-# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
-# segments don't drop below log.retention.bytes.
#log.retention.bytes=1073741824
-
-# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824
-
-# The interval at which log segments are checked to see if they can be deleted according
-# to the retention policies
log.retention.check.interval.ms=300000
-
-# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.
-# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
log.cleaner.enable=false
-
-############################# Zookeeper #############################
-
-# Zookeeper connection string (see zookeeper docs for details).
-# This is a comma separated host:port pairs, each corresponding to a zk
-# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
-# You can also append an optional chroot string to the urls to specify the
-# root directory for all kafka znodes.
zookeeper.connect={{ zk.connect_setting() }}
-
-# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=2000
+auto.create.topics.enable=false
diff --git a/tests/kafkatest/services/verifiable_producer.py b/tests/kafkatest/services/verifiable_producer.py
index cca8227702200..97e626afc154e 100644
--- a/tests/kafkatest/services/verifiable_producer.py
+++ b/tests/kafkatest/services/verifiable_producer.py
@@ -16,23 +16,25 @@
from ducktape.services.background_thread import BackgroundThreadService
import json
-
+import re
class VerifiableProducer(BackgroundThreadService):
logs = {
"producer_log": {
"path": "/mnt/producer.log",
- "collect_default": False}
+ "collect_default": True}
}
- def __init__(self, context, num_nodes, kafka, topic, max_messages=-1, throughput=100000):
+ def __init__(self, context, num_nodes, kafka, topic, max_messages=-1, throughput=100000, close_timeout_ms=None, configs={}):
super(VerifiableProducer, self).__init__(context, num_nodes)
self.kafka = kafka
self.topic = topic
self.max_messages = max_messages
self.throughput = throughput
+ self.close_timeout_ms = close_timeout_ms
+ self.configs = configs # Overrides for Kafka producer settings
self.acked_values = []
self.not_acked_values = []
@@ -48,12 +50,8 @@ def _worker(self, idx, node):
if data is not None:
with self.lock:
- if data["name"] == "producer_send_error":
- data["node"] = idx
- self.not_acked_values.append(int(data["value"]))
-
- elif data["name"] == "producer_send_success":
- self.acked_values.append(int(data["value"]))
+ if "v" in data.keys():
+ self.acked_values.append(int(data["v"]))
@property
def start_cmd(self):
@@ -64,6 +62,12 @@ def start_cmd(self):
if self.throughput > 0:
cmd += " --throughput %s" % str(self.throughput)
+ if self.close_timeout_ms is not None:
+ cmd += " --close-timeout %s" % str(self.close_timeout_ms)
+
+ for key in self.configs:
+ cmd += " --config %s=%s" % (key, str(self.configs[key]))
+
cmd += " 2>> /mnt/producer.log | tee -a /mnt/producer.log &"
return cmd
diff --git a/tests/kafkatest/services/zookeeper.py b/tests/kafkatest/services/zookeeper.py
index 56f46068791dc..8ca38d8e40410 100644
--- a/tests/kafkatest/services/zookeeper.py
+++ b/tests/kafkatest/services/zookeeper.py
@@ -15,7 +15,10 @@
from ducktape.services.service import Service
+from ducktape.utils.util import wait_until
+from concurrent import futures
+import json
import time
@@ -33,6 +36,17 @@ def __init__(self, context, num_nodes):
"""
super(ZookeeperService, self).__init__(context, num_nodes)
+ def start(self):
+ super(ZookeeperService, self).start()
+ if not wait_until(lambda: self.all_alive(), timeout_sec=20, backoff_sec=.5):
+ raise RuntimeError("Timed out waiting for Zookeeper cluster to start.")
+
+ def all_alive(self):
+ for node in self.nodes:
+ if not self.alive(node):
+ return False
+ return True
+
def start_node(self, node):
idx = self.idx(node)
self.logger.info("Starting ZK node %d on %s", idx, node.account.hostname)
@@ -45,11 +59,19 @@ def start_node(self, node):
self.logger.info(config_file)
node.account.create_file("/mnt/zookeeper.properties", config_file)
- node.account.ssh(
- "/opt/kafka/bin/zookeeper-server-start.sh /mnt/zookeeper.properties 1>> %(path)s 2>> %(path)s &"
- % self.logs["zk_log"])
+ cmd = "/opt/kafka/bin/zookeeper-server-start.sh /mnt/zookeeper.properties 1>> %(path)s 2>> %(path)s &" % \
+ self.logs["zk_log"]
- time.sleep(5) # give it some time to start
+ with futures.ThreadPoolExecutor(max_workers=1) as executor:
+ future = executor.submit(lambda: node.account.ssh(cmd))
+
+ def alive(self, node):
+ try:
+ # Try opening tcp connection and immediately closing by sending EOF
+ node.account.ssh("echo EOF | nc %s %d" % (node.account.hostname, 2181), allow_fail=False)
+ return True
+ except:
+ return False
def stop_node(self, node):
idx = self.idx(node)
@@ -62,3 +84,22 @@ def clean_node(self, node):
def connect_setting(self):
return ','.join([node.account.hostname + ':2181' for node in self.nodes])
+
+ def get_data(self, path):
+ """
+ Get data from zookeeper on the given path. This method assumes the data is in JSON format.
+
+ :param path Zookeeper path to query for data
+ """
+ cmd = "/opt/kafka/bin/kafka-run-class.sh kafka.tools.ZooKeeperMainWrapper -server %s get %s" % \
+ (self.connect_setting(), path)
+ data = None
+ node = self.nodes[0]
+ for line in node.account.ssh_capture(cmd):
+ try:
+ data = json.loads(line)
+ break
+ except ValueError:
+ pass
+
+ return data
\ No newline at end of file
diff --git a/tests/kafkatest/tests/replication_test.py b/tests/kafkatest/tests/replication_test.py
index fed1ea1f8e2f3..7c848cc903822 100644
--- a/tests/kafkatest/tests/replication_test.py
+++ b/tests/kafkatest/tests/replication_test.py
@@ -15,45 +15,78 @@
from ducktape.tests.test import Test
from ducktape.utils.util import wait_until
+from ducktape.mark import matrix
+from ducktape.mark import parametrize
from kafkatest.services.zookeeper import ZookeeperService
from kafkatest.services.kafka import KafkaService
from kafkatest.services.verifiable_producer import VerifiableProducer
from kafkatest.services.console_consumer import ConsoleConsumer
-import signal
-import time
+from kafkatest.process_signal import SIGTERM, SIGKILL, SIGSTOP
+
+from concurrent import futures
+
+# Death "permanence"
+BOUNCE = "bounce" # Kill and revive
+KILL = "kill" # Kill without reviving
+
+# node types
+LEADER = "leader"
+CONTROLLER = "controller"
class ReplicationTest(Test):
"""Replication tests.
These tests verify that replication provides simple durability guarantees by checking that data acked by
- brokers is still available for consumption in the face of various failure scenarios."""
+ brokers is still available for consumption in the face of various failure scenarios.
+
+ Each test produces to three separate topics while driving various kinds of failure on a particular topic and partition.
+ This exercises replication and leader/controller failover while producing across multiple topics, each with multiple partitions.
+
+ Failures are induced on either the leader of a topic/partition or on the controller. As a consequence of the
+ way brokers distribute leaders across the broker cluster, this will have the side effect of exercising
+ failure of followers as wel..
+
+ We then validate by consuming all messages from each topic, and checking that every acked message shows up in
+ the downstream consumer for each topic.
+ """
def __init__(self, test_context):
- """:type test_context: ducktape.tests.test.TestContext"""
super(ReplicationTest, self).__init__(test_context=test_context)
- self.topic = "test_topic"
self.zk = ZookeeperService(test_context, num_nodes=1)
- self.kafka = KafkaService(test_context, num_nodes=3, zk=self.zk, topics={self.topic: {
- "partitions": 3,
- "replication-factor": 3,
- "min.insync.replicas": 2}
- })
+
+ # Important to disable unclean leader election
+ topic_config = {"partitions": 3, "replication-factor": 3,
+ "configs": {"min.insync.replicas": 2, "unclean.leader.election.enable": "false"}}
+ self.topic_names = ["topic1", "topic2", "topic3"]
+ self.topics = {topic: topic_config.copy() for topic in self.topic_names}
+ self.topic_to_fail = "topic1" # We'll induce failures on this topic
+
+ self.kafka = KafkaService(test_context, num_nodes=3, zk=self.zk, topics=self.topics)
self.producer_throughput = 10000
- self.num_producers = 1
- self.num_consumers = 1
+
+ self.producers = [VerifiableProducer(self.test_context, 1, self.kafka, topic,
+ throughput=self.producer_throughput,
+ # close_timeout_ms=60000,
+ # configs={"acks": -1})
+ configs={"acks": -1, "batch.size": 100, "buffer.memory": 1000})
+ for topic in self.topic_names]
+ self.consumers = [ConsoleConsumer(self.test_context, 1, self.kafka, topic, consumer_timeout_ms=3000)
+ for topic in self.topic_names]
def setUp(self):
self.zk.start()
self.kafka.start()
- def min_cluster_size(self):
- """Override this since we're adding services outside of the constructor"""
- return super(ReplicationTest, self).min_cluster_size() + self.num_producers + self.num_consumers
-
- def run_with_failure(self, failure):
+ # @parametrize(configs={"acks": "1"})
+ @parametrize(configs={"compression.type": "gzip"})
+ @matrix(failure=[SIGTERM, SIGSTOP, SIGKILL])
+ @matrix(node_type=[LEADER, CONTROLLER])
+ # @matrix(failure_permanence=[BOUNCE, KILL])
+ def test_replication(self, failure=SIGTERM, failure_permanence=BOUNCE,
+ node_type=LEADER, configs={}, num_bounce=4):
"""This is the top-level test template.
The steps are:
@@ -75,91 +108,158 @@ def run_with_failure(self, failure):
indicator that nothing is left to consume.
"""
- self.producer = VerifiableProducer(self.test_context, self.num_producers, self.kafka, self.topic, throughput=self.producer_throughput)
- self.consumer = ConsoleConsumer(self.test_context, self.num_consumers, self.kafka, self.topic, consumer_timeout_ms=3000)
+ self.num_bounce = num_bounce
+
+ for producer in self.producers:
+ producer.configs.update(configs.copy())
# Produce in a background thread while driving broker failures
- self.producer.start()
- if not wait_until(lambda: self.producer.num_acked > 5, timeout_sec=5):
- raise RuntimeError("Producer failed to start in a reasonable amount of time.")
- failure()
- self.producer.stop()
+ self.logger.debug("Producing messages...")
+ self.start_producers()
+
+ self.logger.debug("Driving failures...")
+
+ assert self.kafka.all_alive()
+
+ node_to_signal = self.fetch_broker_node(self.topic_to_fail, 0, node_type)
+ assert node_to_signal != None
+ leader_or_controller_change = \
+ lambda: node_to_signal != self.fetch_broker_node(self.topic_to_fail, 0, node_type) and self.kafka.dead(node_to_signal)
+
+ self.kafka.bounce_node(
+ node_to_signal, sig=SIGTERM, condition=leader_or_controller_change,
+ condition_timeout_sec=30, condition_backoff_sec=.5)
- self.acked = self.producer.acked
- self.not_acked = self.producer.not_acked
- self.logger.info("num not acked: %d" % self.producer.num_not_acked)
- self.logger.info("num acked: %d" % self.producer.num_acked)
+ self.drive_failures(failure, failure_permanence, node_type)
+ self.stop_producers()
+
+ self.acked = [producer.acked for producer in self.producers]
# Consume all messages
- self.consumer.start()
- self.consumer.wait()
- self.consumed = self.consumer.messages_consumed[1]
- self.logger.info("num consumed: %d" % len(self.consumed))
+ self.logger.debug("Consuming messages...")
+ self.messages_consumed = []
+ for consumer in self.consumers:
+ consumer.start()
+ consumer.wait()
+ self.messages_consumed.append(consumer.messages_consumed[1])
# Check produced vs consumed
+ self.logger.debug("Validating...")
success, msg = self.validate()
if not success:
- self.mark_for_collect(self.producer)
+ for producer in self.producers:
+ self.mark_for_collect(producer)
assert success, msg
- def clean_shutdown(self):
- """Discover leader node for our topic and shut it down cleanly."""
- self.kafka.signal_leader(self.topic, partition=0, sig=signal.SIGTERM)
-
- def hard_shutdown(self):
- """Discover leader node for our topic and shut it down with a hard kill."""
- self.kafka.signal_leader(self.topic, partition=0, sig=signal.SIGKILL)
-
- def clean_bounce(self):
- """Chase the leader of one partition and restart it cleanly."""
- for i in range(5):
- prev_leader_node = self.kafka.leader(topic=self.topic, partition=0)
- self.kafka.restart_node(prev_leader_node, wait_sec=5, clean_shutdown=True)
-
- def hard_bounce(self):
- """Chase the leader and restart it cleanly."""
- for i in range(5):
- prev_leader_node = self.kafka.leader(topic=self.topic, partition=0)
- self.kafka.restart_node(prev_leader_node, wait_sec=5, clean_shutdown=False)
-
- # Wait long enough for previous leader to probably be awake again
- time.sleep(6)
+ def fetch_broker_node(self, topic, partition, node_type):
+ if node_type == LEADER:
+ leader = self.kafka.leader(topic=topic, partition=partition)
+ self.logger.info("Leader for topic %s and partition %d is currently: %d" % (topic, partition, self.kafka.idx(leader)))
+ return leader
+ elif node_type == CONTROLLER:
+ controller = self.kafka.controller()
+ self.logger.info("Controller is currently: %d" % (self.kafka.idx(controller)))
+ return controller
+ else:
+ raise RuntimeError("Unsupported node type.")
+
+ def start_producers(self):
+ for producer in self.producers:
+ producer.start()
+
+ for producer in self.producers:
+ if not wait_until(lambda: producer.num_acked > 5, timeout_sec=5):
+ raise RuntimeError("Producer failed to start in a reasonable amount of time: %s" % str(producer))
+
+ def stop_producers(self):
+ stop_futures = []
+ with futures.ThreadPoolExecutor(max_workers=len(self.topic_names)) as executor:
+ for producer in self.producers:
+ stop_futures.append(executor.submit(producer.stop))
+
+ def done():
+ """:return True iff all futures are done."""
+ done_futures = [f.done() for f in stop_futures]
+ return reduce(lambda x, y: x and y, done_futures, True)
+
+ if not wait_until(lambda: done(), timeout_sec=30, backoff_sec=1):
+ raise RuntimeError("Producers did not stop in a reasonable amount of time.")
+
+ def drive_failures(self, failure, failure_permanence, node_type):
+
+ if failure_permanence == KILL:
+ node_to_signal = self.fetch_broker_node(self.topic_to_fail, 0, node_type)
+ self.kafka.signal_node(node_to_signal, sig=failure)
+
+ elif failure_permanence == BOUNCE:
+ self.bounce(failure, node_type, self.num_bounce)
+ else:
+ raise RuntimeError("Invalid failure type")
+
+ def bounce(self, signal, node_type, num_bounce):
+
+ for i in range(num_bounce):
+ assert self.kafka.all_alive()
+ node_to_signal = self.fetch_broker_node(self.topic_to_fail, 0, node_type)
+ self.logger.debug("Will bounce " + str(node_to_signal.account))
+ assert node_to_signal != None
+
+ def leader_or_controller_change():
+ changed = node_to_signal is not None \
+ and node_to_signal != self.fetch_broker_node(self.topic_to_fail, 0, node_type)
+ if signal == SIGSTOP:
+ # Check for deadness doesn't really work on a paused process
+ return changed
+ else:
+ # Make sure the new controller/leader is elected *and* the old process is dead
+ # Otherwise the process will almost certainly fail to restart
+ return changed and self.kafka.dead(node_to_signal)
+
+ self.kafka.bounce_node(
+ node_to_signal, sig=signal, condition=leader_or_controller_change,
+ condition_timeout_sec=30, condition_backoff_sec=.5)
+
+ assert self.kafka.all_alive()
+
+ def kill_permanently(self, failure, node_type):
+ """Fail by sending a signal to the process, but don't revive."""
+ node_to_signal = self.fetch_broker_node(self.topic_to_fail, 0, node_type)
+ self.kafka.signal_node(node_to_signal, failure)
+ num_acked = self.producers[0].num_acked
+
+ # Make sure that writes are eventually able to go through again after the failure
+ wait_until(lambda: self.producers[0].num_acked > num_acked + 100, timeout_sec=6)
def validate(self):
"""Check that produced messages were consumed."""
success = True
msg = ""
-
- if len(set(self.consumed)) != len(self.consumed):
- # There are duplicates. This is ok, so report it but don't fail the test
- msg += "There are duplicate messages in the log\n"
-
- if not set(self.consumed).issuperset(set(self.acked)):
- # Every acked message must appear in the logs. I.e. consumed messages must be superset of acked messages.
- acked_minus_consumed = set(self.producer.acked) - set(self.consumed)
- success = False
- msg += "At least one acked message did not appear in the consumed messages. acked_minus_consumed: " + str(acked_minus_consumed)
+ for i in range(len(self.topic_names)):
+ consumed = self.messages_consumed[i]
+ acked = self.acked[i]
+ topic = self.topic_names[i]
+
+ if len(consumed) == 0:
+ msg += "No messages consumed under topic %s" % topic
+ success = False
+
+ if len(set(consumed)) != len(consumed):
+ # There are duplicates. This is ok, so report it but don't fail the test
+ msg += "There are duplicate messages in the log\n"
+
+ if not set(consumed).issuperset(set(acked)):
+ # Every acked message must appear in the logs. I.e. consumed messages must be superset of acked messages.
+ acked_minus_consumed = set(acked) - set(consumed)
+ success = False
+ msg += "At least one acked message did not appear in the consumed messages for topic %s: " % topic
+ msg += "num acked from topic %s: %d. " % (topic, len(acked))
+ msg += "num consumed from topic %s: %d. " % (topic, len(consumed))
if not success:
# Collect all the data logs if there was a failure
self.mark_for_collect(self.kafka)
return success, msg
-
- def test_clean_shutdown(self):
- self.run_with_failure(self.clean_shutdown)
-
- def test_hard_shutdown(self):
- self.run_with_failure(self.hard_shutdown)
-
- def test_clean_bounce(self):
- self.run_with_failure(self.clean_bounce)
-
- def test_hard_bounce(self):
- self.run_with_failure(self.hard_bounce)
-
-
-
diff --git a/tests/kafkatest/tests/simple_network_partition_test.py b/tests/kafkatest/tests/simple_network_partition_test.py
new file mode 100644
index 0000000000000..126ba37c7df52
--- /dev/null
+++ b/tests/kafkatest/tests/simple_network_partition_test.py
@@ -0,0 +1,218 @@
+# Copyright 2015 Confluent Inc.
+#
+# Licensed 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.
+
+
+# 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.
+
+from ducktape.tests.test import Test
+from ducktape.utils.util import wait_until
+
+from kafkatest.services.zookeeper import ZookeeperService
+from kafkatest.services.kafka import KafkaService
+from kafkatest.services.verifiable_producer import VerifiableProducer
+from kafkatest.services.console_consumer import ConsoleConsumer
+
+import time
+
+from kafkatest.process_signal import SIGSTOP, SIGCONT, SIGKILL
+from concurrent import futures
+
+# Death "permanence"
+BOUNCE = "bounce" # Kill and revive
+KILL = "kill" # Kill without reviving
+
+# node types
+LEADER = "leader"
+CONTROLLER = "controller"
+
+
+class SimpleNetworkPartitionTest(Test):
+ """
+ Somewhat ad-hoc network partition test. This roughly reproduces the failure mode outlined here:
+ https://aphyr.com/posts/293-call-me-maybe-kafka
+
+ Setup
+ - 2 Kafka brokers
+ - 1 topic with replication-factor 2 and min.insync.replicas=1
+ - Producer with required acks = 1
+
+ Test
+ - Find leader and follower for topic
+ - Start producer
+ - After some time, partition leader and follower
+ - Continue writes so that follower falls behind
+ - Kill leader
+ - Continue writes
+ - Verify that some of the writes were lost
+
+ """
+
+ def __init__(self, test_context):
+ super(SimpleNetworkPartitionTest, self).__init__(test_context=test_context)
+
+ # Kafka and Zookeeper
+ self.zk = ZookeeperService(test_context, num_nodes=1)
+ topic_config = {"partitions": 1, "replication-factor": 2, "configs": {"min.insync.replicas": 1}}
+ self.topic = "my_topic"
+ self.kafka = KafkaService(test_context, num_nodes=2, zk=self.zk, topics={self.topic: topic_config})
+
+ # Producer and Consumer
+ self.producer_throughput = 1000
+ self.producer = VerifiableProducer(self.test_context, 1, self.kafka, self.topic,
+ throughput=self.producer_throughput,
+ configs={"acks": 1, "batch.size": 1000, "buffer.memory": 10000})
+ self.consumer = ConsoleConsumer(self.test_context, 1, self.kafka, self.topic, consumer_timeout_ms=3000)
+
+ self.leader = None
+ self.follower = None
+
+ def setUp(self):
+ self.zk.start()
+ self.kafka.start()
+
+ def test_replication(self):
+
+ try:
+ self.leader = self.kafka.leader(self.topic, 0)
+ leader_idx = self.kafka.idx(self.leader)
+ follower_idx = 1 + leader_idx % 2 # 1 -> 2, and 2 -> 1
+ self.follower = self.kafka.get_node(follower_idx)
+
+ # Immediately partition leader from follower
+ self.partition_follower_leader(self.follower, self.leader)
+
+
+ # self.kafka.stop_node(self.follower)
+ # self.kafka.signal_node(self.follower, sig=SIGSTOP)
+
+ self.producer.start()
+
+ # Produce for a little while (say, 10,000 messages)
+ # Perhaps check directly that follower has fallen out of isr
+ # Could do this by.... parsing the describe topic command ;)
+ time.sleep(5)
+
+ # Kill leader and do not revive (clean kill is fine)
+ leader_pid = self.kafka.pids(self.leader)[0]
+ self.kafka.signal_node(self.leader, sig=SIGKILL)
+ wait_until(lambda: not self.leader.account.alive(leader_pid), timeout_sec=3, backoff_sec=.25)
+
+ # self.kafka.signal_node(self.follower, sig=SIGCONT)
+ # self.kafka.start_node(self.follower)
+ if not wait_until(lambda: self.follower == self.kafka.leader(self.topic, 0), timeout_sec=10, backoff_sec=.5):
+ raise RuntimeError("Leader reelection did not take place in a reasonable amount of time.")
+
+ acked = self.producer.num_acked
+ self.logger.info("Num currently acked: " + str(acked))
+ self.logger.info("Waiting for more messages to be acked...")
+ if not wait_until(lambda: self.producer.num_acked > acked + 1000, timeout_sec=60, backoff_sec=.5):
+ raise RuntimeError("Former follower is now leader, but new messages are not being acknowledged.")
+ self.stop_producer()
+
+ # consume
+ self.consumer.start()
+ self.consumer.wait()
+ self.messages_consumed = self.consumer.messages_consumed[1]
+
+ if len(self.messages_consumed) == 0:
+ raise RuntimeError("No messages consumed under topic %s" % self.topic)
+
+ # Check produced vs consumed
+ # validation should fail
+ self.logger.debug("Validating...")
+ success, msg = self.validate()
+ assert not success, "Somehow every acked message was consumed despite the network partition."
+ self.logger.info("Test failed in the *expected* way:" + msg)
+ except BaseException as e:
+ raise e
+ finally:
+ self.logger.info("follower: " + str(self.follower))
+ if self.follower and self.follower.account:
+ # self.follower.account.ssh("sudo iptables -D INPUT -p tcp --source %s -j DROP" %
+ # self.leader.account.hostname, allow_fail=True)
+ self.clear_iptables(self.follower)
+ if self.leader and self.leader.account:
+ # self.leader.account.ssh("sudo iptables -D INPUT -p tcp --source %s -j DROP" %
+ # self.follower.account.hostname, allow_fail=True)
+ self.clear_iptables(self.leader)
+
+ def partition_follower_leader(self, follower, leader):
+ follower.account.ssh("sudo iptables -A INPUT -p tcp --source %s -j DROP" %
+ leader.account.hostname)
+ # leader.account.ssh("sudo iptables -A INPUT -p tcp --source %s -j DROP" %
+ # follower.account.hostname)
+
+ def start_producer(self):
+ self.producer.start()
+
+ if not wait_until(lambda: self.producer.num_acked > 5, timeout_sec=5):
+ raise RuntimeError("Producer failed to start in a reasonable amount of time: %s" % str(self.producer))
+
+ def stop_producer(self):
+ with futures.ThreadPoolExecutor(max_workers=1) as executor:
+ future = (executor.submit(self.producer.stop))
+
+ if not wait_until(future.done, timeout_sec=60, backoff_sec=1):
+ raise RuntimeError("Producer did not stop in a reasonable amount of time.")
+
+ def validate(self):
+ """Check that produced messages were consumed."""
+
+ success = True
+ msg = ""
+ consumed = self.messages_consumed
+ acked = self.producer.acked
+
+ if len(set(consumed)) != len(consumed):
+ # There are duplicates. This is ok, so report it but don't fail the test
+ msg += "There are duplicate messages in the log\n"
+
+ if not set(consumed).issuperset(set(acked)):
+ # Every acked message must appear in the logs. I.e. consumed messages must be superset of acked messages.
+ acked_minus_consumed = set(acked) - set(consumed)
+ success = False
+ msg += "At least one acked message did not appear in the consumed messages for topic %s. acked_minus_consumed: %s" % (self.topic, str(acked_minus_consumed))
+
+ self.logger.info("num consumed: " + str(len(consumed)))
+ self.logger.info("num acked: " + str(len(acked)))
+ return success, msg
+
+ def clear_iptables(self, node):
+ cmds = ["iptables -F",
+ "iptables -X",
+ "iptables -t nat -F",
+ "iptables -t nat -X",
+ "iptables -t mangle -F",
+ "iptables -t mangle -X",
+ "iptables -P INPUT ACCEPT",
+ "iptables -P FORWARD ACCEPT",
+ "iptables -P OUTPUT ACCEPT"]
+
+ cmds = ["sudo " + cmd for cmd in cmds]
+ node.account.ssh("; ".join(cmds), allow_fail=True)
+
+ # lines = node.account.ssh_capture("sudo iptables -L")
diff --git a/tests/setup.py b/tests/setup.py
index 5ce4bb797aae5..368d0ca03ad3e 100644
--- a/tests/setup.py
+++ b/tests/setup.py
@@ -23,5 +23,5 @@
platforms=["any"],
license="apache2.0",
packages=find_packages(),
- requires=["ducktape(>=0.2.0)"]
+ requires=["ducktape(==0.2.0)", "futures(==3.0.3)"]
)
diff --git a/tools/src/main/java/org/apache/kafka/clients/tools/VerifiableProducer.java b/tools/src/main/java/org/apache/kafka/clients/tools/VerifiableProducer.java
index 204166a25e740..e5f770242f7cc 100644
--- a/tools/src/main/java/org/apache/kafka/clients/tools/VerifiableProducer.java
+++ b/tools/src/main/java/org/apache/kafka/clients/tools/VerifiableProducer.java
@@ -24,17 +24,20 @@
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
import java.io.IOException;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
+import java.util.List;
import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.impl.Arguments;
import net.sourceforge.argparse4j.inf.ArgumentParser;
import net.sourceforge.argparse4j.inf.ArgumentParserException;
import net.sourceforge.argparse4j.inf.Namespace;
@@ -56,28 +59,55 @@ public class VerifiableProducer {
String topic;
private Producer producer;
- // If maxMessages < 0, produce until the process is killed externally
- private long maxMessages = -1;
+ // Print more detailed information if true
+ private final boolean verbose;
+
+ // If maxMessages < 0, produce until the process is killed externally
+ private final long maxMessages;
+
+ // Throttle message throughput if this is set >= 0
+ private final long throughput;
+
+ // Timeout on producer.close() call
+ private final long closeTimeoutMs;
+
+ private Properties producerProps;
+
// Number of messages for which acks were received
private long numAcked = 0;
// Number of send attempts
private long numSent = 0;
-
- // Throttle message throughput if this is set >= 0
- private long throughput;
-
+
// Hook to trigger producing thread to stop sending messages
- private boolean stopProducing = false;
+ private AtomicBoolean stopProducing = new AtomicBoolean(false);
+
+ // Track when this.close() is called
+ private long timeOfCloseMs;
public VerifiableProducer(
- Properties producerProps, String topic, int throughput, int maxMessages) {
+ Properties producerProps, String topic, int throughput,
+ int maxMessages, long closeTimeoutMs, boolean verbose) {
this.topic = topic;
this.throughput = throughput;
this.maxMessages = maxMessages;
- this.producer = new KafkaProducer(producerProps);
+ this.closeTimeoutMs = closeTimeoutMs;
+ this.verbose = verbose;
+ this.producerProps = producerProps;
+ this.producer = new KafkaProducer<>(producerProps);
+ }
+
+ @Override
+ public String toString() {
+ return this.getClass() +
+ "(producerProps=" + producerProps + ", " +
+ "topic=" + topic + ", " +
+ "throughput=" + throughput + ", " +
+ "maxMessages=" + maxMessages + ", " +
+ "closeTimeoutMs=" + closeTimeoutMs + ", " +
+ "verbose=" + verbose + ")";
}
/** Get the command-line argument parser. */
@@ -100,7 +130,8 @@ private static ArgumentParser argParser() {
.type(String.class)
.metavar("HOST1:PORT1[,HOST2:PORT2[...]]")
.dest("brokerList")
- .help("Comma-separated list of Kafka brokers in the form HOST1:PORT1,HOST2:PORT2,...");
+ .help(
+ "Comma-separated list of Kafka brokers in the form HOST1:PORT1,HOST2:PORT2,...");
parser.addArgument("--max-messages")
.action(store())
@@ -109,7 +140,8 @@ private static ArgumentParser argParser() {
.type(Integer.class)
.metavar("MAX-MESSAGES")
.dest("maxMessages")
- .help("Produce this many messages. If -1, produce messages until the process is killed externally.");
+ .help(
+ "Produce this many messages. If -1, produce messages until the process is killed externally.");
parser.addArgument("--throughput")
.action(store())
@@ -117,16 +149,46 @@ private static ArgumentParser argParser() {
.setDefault(-1)
.type(Integer.class)
.metavar("THROUGHPUT")
- .help("If set >= 0, throttle maximum message throughput to *approximately* THROUGHPUT messages/sec.");
+ .help(
+ "If set >= 0, throttle maximum message throughput to *approximately* THROUGHPUT messages/sec.");
- parser.addArgument("--acks")
+ parser.addArgument("--request-required-acks")
.action(store())
.required(false)
.setDefault(-1)
.type(Integer.class)
.choices(0, 1, -1)
- .metavar("ACKS")
- .help("Acks required on each produced message. See Kafka docs on request.required.acks for details.");
+ .metavar("REQUEST-REQUIRED-ACKS")
+ .dest("acks")
+ .help(
+ "Acks required on each produced message. See Kafka docs on request.required.acks for details.");
+
+ parser.addArgument("--config")
+ .action(Arguments.append())
+ .required(false)
+ .setDefault(new ArrayList())
+ .type(String.class)
+ .dest("userSpecifiedConfigs")
+ .metavar("CONFIG")
+ .help("Any custom producer properties of the form key=val.");
+
+ parser.addArgument("--close-timeout-ms")
+ .action(store())
+ .required(false)
+ .setDefault(Long.MAX_VALUE)
+ .type(Long.class)
+ .metavar("CLOSE-TIMEOUT-MS")
+ .dest("closeTimeoutMs")
+ .help(
+ "When SIGTERM is caught, wait at most this many milliseconds for unsent messages to flush before stopping the VerifiableProducer process.");
+
+ parser.addArgument("--verbose")
+ .action(storeTrue())
+ .required(false)
+ .type(Boolean.class)
+ .metavar("VERBOSE")
+ .dest("verbose")
+ .help("");
return parser;
}
@@ -137,13 +199,13 @@ public static VerifiableProducer createFromArgs(String[] args) {
VerifiableProducer producer = null;
try {
- Namespace res;
- res = parser.parseArgs(args);
-
+ Namespace res = parser.parseArgs(args);
int maxMessages = res.getInt("maxMessages");
String topic = res.getString("topic");
int throughput = res.getInt("throughput");
-
+ long closeTimeoutMs = res.getLong("closeTimeoutMs");
+ boolean verbose = res.getBoolean("verbose");
+
Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, res.getString("brokerList"));
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
@@ -152,9 +214,26 @@ public static VerifiableProducer createFromArgs(String[] args) {
"org.apache.kafka.common.serialization.StringSerializer");
producerProps.put(ProducerConfig.ACKS_CONFIG, Integer.toString(res.getInt("acks")));
// No producer retries
- producerProps.put("retries", "0");
+ producerProps.put(ProducerConfig.RETRIES_CONFIG, "0");
+
+ // Properties specified using the --config key=val format override
+ // other properties
+ List userSpecifiedConfigs = res.getList("userSpecifiedConfigs");
+ System.out.println(userSpecifiedConfigs);
+ for (String config: userSpecifiedConfigs) {
+ int index = config.indexOf('=');
+ if (index < 0 || index == config.length() - 1) {
+ throw new IllegalArgumentException(
+ "Invalid format for producer config: " + config);
+ }
+
+ String key = config.substring(0, index);
+ String val = config.substring(index + 1, config.length());
+ producerProps.put(key, val);
+ }
- producer = new VerifiableProducer(producerProps, topic, throughput, maxMessages);
+ producer = new VerifiableProducer(
+ producerProps, topic, throughput, maxMessages, closeTimeoutMs, verbose);
} catch (ArgumentParserException e) {
if (args.length == 0) {
parser.printHelp();
@@ -175,65 +254,109 @@ public void send(String key, String value) {
try {
producer.send(record, new PrintInfoCallback(key, value));
} catch (Exception e) {
-
synchronized (System.out) {
- System.out.println(errorString(e, key, value, System.currentTimeMillis()));
+ reportError(e, key, value);
+
}
}
}
+ private void reportSuccess(RecordMetadata recordMetadata, String key, String value) {
+ if (verbose) {
+ System.out.println(
+ successString(recordMetadata, key, value, System.currentTimeMillis()));
+ } else {
+ String data = "{\"p\":" + recordMetadata.partition();
+ // Only add non-empty key
+ data += (key == null || key.length() == 0) ? "" : ",\"k\":\"" + key + "\"";
+ data += ",\"v\":\"" + value + "\"}";
+
+ System.out.println(data);
+ }
+ }
+
+ /** Report send error if in verbose mode. */
+ private void reportError(Exception e, String key, String value) {
+ if (verbose) {
+ System.out.println(errorString(e, key, value, System.currentTimeMillis()));
+ }
+ }
+
/** Close the producer to flush any remaining messages. */
public void close() {
- producer.close();
+ // Trigger main thread to stop producing messages
+ stopProducing.set(true);
+ timeOfCloseMs = System.currentTimeMillis();
+ producer.close(this.closeTimeoutMs, TimeUnit.MILLISECONDS);
}
-
+
/**
- * Return JSON string encapsulating basic information about the exception, as well
- * as the key and value which triggered the exception.
+ * Helper method used in constructing JSON strings.
+ * Although using a JSON library is more elegant, it's also significantly slower than
+ * directly creating the small strings we need for this tool.
*/
- String errorString(Exception e, String key, String value, Long nowMs) {
- assert e != null : "Expected non-null exception.";
-
- Map errorData = new HashMap<>();
- errorData.put("class", this.getClass().toString());
- errorData.put("name", "producer_send_error");
+ private static StringBuilder keyValToString(String key, String val) {
+ StringBuilder builder = new StringBuilder();
+ builder.append("\"");
+ builder.append(key);
+ builder.append("\":");
- errorData.put("time_ms", nowMs);
- errorData.put("exception", e.getClass().toString());
- errorData.put("message", e.getMessage());
- errorData.put("topic", this.topic);
- errorData.put("key", key);
- errorData.put("value", value);
+ builder.append("\"");
+ builder.append(val);
+ builder.append("\"");
+ return builder;
+ }
+
+ private static String toJsonString(Map map) {
+ StringBuilder builder = new StringBuilder();
- return toJsonString(errorData);
+ builder.append("{");
+ int i = 0;
+ for (String key: map.keySet()) {
+ builder.append(keyValToString(key, String.valueOf(map.get(key))));
+ if (i < map.keySet().size() - 1) {
+ builder.append(",");
+ }
+ i++;
+ }
+ builder.append("}");
+ return builder.toString();
}
String successString(RecordMetadata recordMetadata, String key, String value, Long nowMs) {
assert recordMetadata != null : "Expected non-null recordMetadata object.";
- Map successData = new HashMap<>();
- successData.put("class", this.getClass().toString());
- successData.put("name", "producer_send_success");
+ Map map = new HashMap<>();
+ map.put("class", this.getClass().toString());
+ map.put("name", "producer_send_success");
+ map.put("time_ms", nowMs);
+ map.put("topic", this.topic);
+ map.put("partition", String.valueOf(recordMetadata.partition()));
+ map.put("offset", String.valueOf(recordMetadata.offset()));
+ map.put("key", key);
+ map.put("value", value);
- successData.put("time_ms", nowMs);
- successData.put("topic", this.topic);
- successData.put("partition", recordMetadata.partition());
- successData.put("offset", recordMetadata.offset());
- successData.put("key", key);
- successData.put("value", value);
-
- return toJsonString(successData);
+ return toJsonString(map);
}
-
- private String toJsonString(Map data) {
- String json;
- try {
- ObjectMapper mapper = new ObjectMapper();
- json = mapper.writeValueAsString(data);
- } catch(JsonProcessingException e) {
- json = "Bad data can't be written as json: " + e.getMessage();
- }
- return json;
+
+ /**
+ * Return JSON string encapsulating basic information about the exception, as well
+ * as the key and value which triggered the exception.
+ */
+ String errorString(Exception e, String key, String value, Long nowMs) {
+ assert e != null : "Expected non-null exception.";
+
+ Map map = new HashMap<>();
+ map.put("class", this.getClass().toString());
+ map.put("name", "producer_send_error");
+ map.put("time_ms", nowMs);
+ map.put("exception", e.getClass());
+ map.put("message", e.getMessage());
+ map.put("topic", this.topic);
+ map.put("key", String.valueOf(key));
+ map.put("value", String.valueOf(value));
+
+ return toJsonString(map);
}
/** Callback which prints errors to stdout when the producer fails to send. */
@@ -248,12 +371,13 @@ private class PrintInfoCallback implements Callback {
}
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
+
synchronized (System.out) {
if (e == null) {
VerifiableProducer.this.numAcked++;
- System.out.println(successString(recordMetadata, this.key, this.value, System.currentTimeMillis()));
+ reportSuccess(recordMetadata, key, value);
} else {
- System.out.println(errorString(e, this.key, this.value, System.currentTimeMillis()));
+ reportError(e, key, value);
}
}
}
@@ -262,37 +386,38 @@ public void onCompletion(RecordMetadata recordMetadata, Exception e) {
public static void main(String[] args) throws IOException {
final VerifiableProducer producer = createFromArgs(args);
+ System.out.println(producer);
+
final long startMs = System.currentTimeMillis();
boolean infinite = producer.maxMessages < 0;
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
- // Trigger main thread to stop producing messages
- producer.stopProducing = true;
-
+ System.out.println("CAUGHT SIGTERM!! FLUSHING BUFFERED MESSAGES...");
+
// Flush any remaining messages
producer.close();
+ long stopMs = System.currentTimeMillis();
// Print a summary
- long stopMs = System.currentTimeMillis();
double avgThroughput = 1000 * ((producer.numAcked) / (double) (stopMs - startMs));
-
- Map data = new HashMap<>();
- data.put("class", producer.getClass().toString());
- data.put("name", "tool_data");
- data.put("sent", producer.numSent);
- data.put("acked", producer.numAcked);
- data.put("target_throughput", producer.throughput);
- data.put("avg_throughput", avgThroughput);
-
- System.out.println(producer.toJsonString(data));
+
+ Map map = new HashMap<>();
+ map.put("class", this.getClass().toString());
+ map.put("name", "tool_data");
+ map.put("sent", producer.numSent);
+ map.put("acked", producer.numAcked);
+ map.put("target_throughput", producer.throughput);
+ map.put("avg_throughput", avgThroughput);
+
+ System.out.println(toJsonString(map));
}
});
ThroughputThrottler throttler = new ThroughputThrottler(producer.throughput, startMs);
for (int i = 0; i < producer.maxMessages || infinite; i++) {
- if (producer.stopProducing) {
+ if (producer.stopProducing.get()) {
break;
}
long sendStartMs = System.currentTimeMillis();
diff --git a/vagrant/base.sh b/vagrant/base.sh
index 133f10a95622c..0968726d30a40 100644
--- a/vagrant/base.sh
+++ b/vagrant/base.sh
@@ -22,6 +22,14 @@ if [ -z `which javac` ]; then
apt-get install -y software-properties-common python-software-properties
add-apt-repository -y ppa:webupd8team/java
apt-get -y update
+
+ # Need this so that iptables can be used (e.g. for creating network partitions)
+ # Can't use grep here because of exit status; vagrant provisioning will halt if exit status is non-zero
+ s=`awk '/ip_tables/{ print $0 }'`
+ if [ "x$s" == "x" ]; then
+ echo "INSERTING MODULE"
+ echo "ip_tables" >> /etc/modules
+ fi
# Try to share cache. See Vagrantfile for details
mkdir -p /var/cache/oracle-jdk7-installer
@@ -38,10 +46,12 @@ if [ -z `which javac` ]; then
fi
chmod a+rw /opt
-if [ ! -e /opt/kafka ]; then
- ln -s /vagrant /opt/kafka
+if [ -h /opt/kafka ]; then
+ rm /opt/kafka
fi
+ln -s /vagrant /opt/kafka
+
# For EC2 nodes, we want to use /mnt, which should have the local disk. On local
# VMs, we can just create it if it doesn't exist and use it like we'd use
# /tmp. Eventually, we'd like to also support more directories, e.g. when EC2