Repository: storm Updated Branches: refs/heads/master 577e34e7e -> 5383ac375
storm-826, update KafkaBolt to use the new kafka producer APIs Deal with the producer cannot close issue Able to pass unit tests and add callback async Update the producer api in KafkaUtilsTest. Address comments and add test options for async and fireAndForget Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9319ab13 Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9319ab13 Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9319ab13 Branch: refs/heads/master Commit: 9319ab137efa9d606e96d1fc0504502bbaec87e6 Parents: 28e64c9 Author: zhuol <[email protected]> Authored: Wed May 27 10:48:39 2015 -0500 Committer: zhuol <[email protected]> Committed: Fri Jun 26 10:56:57 2015 -0500 ---------------------------------------------------------------------- external/storm-kafka/pom.xml | 10 ++- .../src/jvm/storm/kafka/bolt/KafkaBolt.java | 77 +++++++++++++++++--- .../src/test/storm/kafka/KafkaUtilsTest.java | 32 +++++--- .../test/storm/kafka/bolt/KafkaBoltTest.java | 74 ++++++++++++++++--- 4 files changed, 157 insertions(+), 36 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/9319ab13/external/storm-kafka/pom.xml ---------------------------------------------------------------------- diff --git a/external/storm-kafka/pom.xml b/external/storm-kafka/pom.xml index 0aba519..c731a0f 100644 --- a/external/storm-kafka/pom.xml +++ b/external/storm-kafka/pom.xml @@ -113,8 +113,8 @@ </dependency> <dependency> <groupId>org.apache.kafka</groupId> - <artifactId>kafka_2.9.2</artifactId> - <version>0.8.1.1</version> + <artifactId>kafka_2.10</artifactId> + <version>0.8.2.1</version> <!-- use provided scope, so users can pull in whichever scala version they choose --> <scope>provided</scope> <exclusions> @@ -129,6 +129,12 @@ </exclusions> </dependency> <dependency> + <groupId>org.apache.kafka</groupId> + <artifactId>kafka-clients</artifactId> + <version>0.8.2.1</version> + <scope>provided</scope> + </dependency> + <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-core</artifactId> <version>${project.version}</version> http://git-wip-us.apache.org/repos/asf/storm/blob/9319ab13/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java b/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java index 714ecd3..738b358 100644 --- a/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java +++ b/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java @@ -23,16 +23,18 @@ import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.base.BaseRichBolt; import backtype.storm.tuple.Tuple; import backtype.storm.utils.TupleUtils; -import kafka.javaapi.producer.Producer; -import kafka.producer.KeyedMessage; -import kafka.producer.ProducerConfig; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.clients.producer.Callback; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper; import storm.kafka.bolt.mapper.TupleToKafkaMapper; import storm.kafka.bolt.selector.DefaultTopicSelector; import storm.kafka.bolt.selector.KafkaTopicSelector; - +import java.util.concurrent.Future; +import java.util.concurrent.ExecutionException; import java.util.Map; import java.util.Properties; @@ -45,6 +47,10 @@ import java.util.Properties; * 'kafka.broker.properties' and 'topic' * <p/> * respectively. + * <p/> + * This bolt uses 0.8.2 Kafka Producer API. + * <p/> + * It works for sending tuples to older Kafka version (0.8.1). */ public class KafkaBolt<K, V> extends BaseRichBolt { @@ -53,10 +59,18 @@ public class KafkaBolt<K, V> extends BaseRichBolt { public static final String TOPIC = "topic"; public static final String KAFKA_BROKER_PROPERTIES = "kafka.broker.properties"; - private Producer<K, V> producer; + private KafkaProducer<K, V> producer; private OutputCollector collector; private TupleToKafkaMapper<K,V> mapper; private KafkaTopicSelector topicSelector; + /** + * With default setting for fireAndForget and async, the callback is called when the sending succeeds. + * By setting fireAndForget true, the send will not wait at all for kafka to ack. + * "acks" setting in 0.8.2 Producer API config doesn't matter if fireAndForget is set. + * By setting async false, synchronous sending is used. + */ + private boolean fireAndForget = false; + private boolean async = true; public KafkaBolt<K,V> withTupleToKafkaMapper(TupleToKafkaMapper<K,V> mapper) { this.mapper = mapper; @@ -83,18 +97,16 @@ public class KafkaBolt<K, V> extends BaseRichBolt { Map configMap = (Map) stormConf.get(KAFKA_BROKER_PROPERTIES); Properties properties = new Properties(); properties.putAll(configMap); - ProducerConfig config = new ProducerConfig(properties); - producer = new Producer<K, V>(config); + producer = new KafkaProducer<K, V>(properties); this.collector = collector; } @Override - public void execute(Tuple input) { + public void execute(final Tuple input) { if (TupleUtils.isTick(input)) { collector.ack(input); return; // Do not try to send ticks to Kafka } - K key = null; V message = null; String topic = null; @@ -102,12 +114,40 @@ public class KafkaBolt<K, V> extends BaseRichBolt { key = mapper.getKeyFromTuple(input); message = mapper.getMessageFromTuple(input); topic = topicSelector.getTopic(input); - if(topic != null ) { - producer.send(new KeyedMessage<K, V>(topic, key, message)); + if (topic != null ) { + Callback callback = null; + + if (!fireAndForget && async) { + callback = new Callback() { + @Override + public void onCompletion(RecordMetadata ignored, Exception e) { + synchronized (collector) { + if (e != null) { + collector.reportError(e); + collector.fail(input); + } else { + collector.ack(input); + } + } + } + }; + } + Future<RecordMetadata> result = producer.send(new ProducerRecord<K, V>(topic, key, message), callback); + if (!async) { + try { + result.get(); + collector.ack(input); + } catch (ExecutionException err) { + collector.reportError(err); + collector.fail(input); + } + } else if (fireAndForget) { + collector.ack(input); + } } else { LOG.warn("skipping key = " + key + ", topic selector returned null."); + collector.ack(input); } - collector.ack(input); } catch (Exception ex) { collector.reportError(ex); collector.fail(input); @@ -118,4 +158,17 @@ public class KafkaBolt<K, V> extends BaseRichBolt { public void declareOutputFields(OutputFieldsDeclarer declarer) { } + + @Override + public void cleanup() { + producer.close(); + } + + public void setFireAndForget(boolean fireAndForget) { + this.fireAndForget = fireAndForget; + } + + public void setAsync(boolean async) { + this.async = async; + } } http://git-wip-us.apache.org/repos/asf/storm/blob/9319ab13/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java index 965eaea..a2d2af8 100644 --- a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java +++ b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java @@ -23,13 +23,14 @@ import com.google.common.collect.ImmutableMap; import kafka.api.OffsetRequest; import kafka.javaapi.consumer.SimpleConsumer; import kafka.javaapi.message.ByteBufferMessageSet; -import kafka.javaapi.producer.Producer; import kafka.message.MessageAndOffset; -import kafka.producer.KeyedMessage; -import kafka.producer.ProducerConfig; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.junit.Assert; import storm.kafka.trident.GlobalPartitionInformation; import java.util.List; @@ -39,9 +40,11 @@ import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.CoreMatchers.is; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertThat; - +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class KafkaUtilsTest { + private static final Logger LOG = LoggerFactory.getLogger(KafkaUtilsTest.class); private KafkaTestBroker broker; private SimpleConsumer simpleConsumer; private KafkaConfig config; @@ -182,7 +185,6 @@ public class KafkaUtilsTest { } } - private void createTopicAndSendMessage() { createTopicAndSendMessage(null, "someValue"); } @@ -193,14 +195,22 @@ public class KafkaUtilsTest { private void createTopicAndSendMessage(String key, String value) { Properties p = new Properties(); - p.setProperty("metadata.broker.list", broker.getBrokerConnectionString()); - p.setProperty("serializer.class", "kafka.serializer.StringEncoder"); - ProducerConfig producerConfig = new ProducerConfig(p); - Producer<String, String> producer = new Producer<String, String>(producerConfig); - producer.send(new KeyedMessage<String, String>(config.topic, key, value)); + p.put("serializer.class", "kafka.serializer.StringEncoder"); + p.put("bootstrap.servers", broker.getBrokerConnectionString()); + p.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + p.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + p.put("metadata.fetch.timeout.ms", 1000); + KafkaProducer<String, String> producer = new KafkaProducer<String, String>(p); + try { + producer.send(new ProducerRecord<String, String>(config.topic, key, value)).get(); + } catch (Exception e) { + Assert.fail(e.getMessage()); + LOG.error("Failed to do synchronous sending due to " + e, e); + } finally { + producer.close(); + } } - @Test public void assignOnePartitionPerTask() { runPartitionToTaskMappingTest(16, 1); http://git-wip-us.apache.org/repos/asf/storm/blob/9319ab13/external/storm-kafka/src/test/storm/kafka/bolt/KafkaBoltTest.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka/src/test/storm/kafka/bolt/KafkaBoltTest.java b/external/storm-kafka/src/test/storm/kafka/bolt/KafkaBoltTest.java index 576cc12..05d138b 100644 --- a/external/storm-kafka/src/test/storm/kafka/bolt/KafkaBoltTest.java +++ b/external/storm-kafka/src/test/storm/kafka/bolt/KafkaBoltTest.java @@ -34,9 +34,7 @@ import kafka.javaapi.consumer.SimpleConsumer; import kafka.javaapi.message.ByteBufferMessageSet; import kafka.message.Message; import kafka.message.MessageAndOffset; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; +import org.junit.*; import org.mockito.Mock; import org.mockito.MockitoAnnotations; import storm.kafka.*; @@ -77,9 +75,9 @@ public class KafkaBoltTest { public void shutdown() { simpleConsumer.close(); broker.shutdown(); + bolt.cleanup(); } - private void setupKafkaConsumer() { GlobalPartitionInformation globalPartitionInformation = new GlobalPartitionInformation(); globalPartitionInformation.addPartition(0, Broker.fromString(broker.getBrokerConnectionString())); @@ -110,15 +108,60 @@ public class KafkaBoltTest { verifyMessage(key, message); } + /* test synchronous sending */ + @Test + public void executeWithByteArrayKeyAndMessageSync() { + boolean async = false; + boolean fireAndForget = false; + bolt = generateDefaultSerializerBolt(async, fireAndForget); + String keyString = "test-key"; + String messageString = "test-message"; + byte[] key = keyString.getBytes(); + byte[] message = messageString.getBytes(); + Tuple tuple = generateTestTuple(key, message); + bolt.execute(tuple); + verify(collector).ack(tuple); + verifyMessage(keyString, messageString); + } + + /* test asynchronous sending (default) */ + @Test + public void executeWithByteArrayKeyAndMessageAsync() { + boolean async = true; + boolean fireAndForget = false; + bolt = generateDefaultSerializerBolt(async, fireAndForget); + String keyString = "test-key"; + String messageString = "test-message"; + byte[] key = keyString.getBytes(); + byte[] message = messageString.getBytes(); + Tuple tuple = generateTestTuple(key, message); + bolt.execute(tuple); + try { + Thread.sleep(1000); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + verify(collector).ack(tuple); + verifyMessage(keyString, messageString); + } + + /* test with fireAndForget option enabled */ @Test - public void executeWithByteArrayKeyAndMessage() { - bolt = generateDefaultSerializerBolt(); + public void executeWithByteArrayKeyAndMessageFire() { + boolean async = true; + boolean fireAndForget = true; + bolt = generateDefaultSerializerBolt(async, fireAndForget); String keyString = "test-key"; String messageString = "test-message"; byte[] key = keyString.getBytes(); byte[] message = messageString.getBytes(); Tuple tuple = generateTestTuple(key, message); bolt.execute(tuple); + try { + Thread.sleep(1000); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } verify(collector).ack(tuple); verifyMessage(keyString, messageString); } @@ -126,21 +169,32 @@ public class KafkaBoltTest { private KafkaBolt generateStringSerializerBolt() { KafkaBolt bolt = new KafkaBolt(); Properties props = new Properties(); - props.put("metadata.broker.list", broker.getBrokerConnectionString()); props.put("request.required.acks", "1"); props.put("serializer.class", "kafka.serializer.StringEncoder"); + props.put("bootstrap.servers", broker.getBrokerConnectionString()); + props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + props.put("metadata.fetch.timeout.ms", 1000); config.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, props); bolt.prepare(config, null, new OutputCollector(collector)); + bolt.setAsync(false); return bolt; } - private KafkaBolt generateDefaultSerializerBolt() { + private KafkaBolt generateDefaultSerializerBolt(boolean async, boolean fireAndForget) { KafkaBolt bolt = new KafkaBolt(); Properties props = new Properties(); - props.put("metadata.broker.list", broker.getBrokerConnectionString()); props.put("request.required.acks", "1"); + props.put("serializer.class", "kafka.serializer.StringEncoder"); + props.put("bootstrap.servers", broker.getBrokerConnectionString()); + props.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); + props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); + props.put("metadata.fetch.timeout.ms", 1000); + props.put("linger.ms", 0); config.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, props); bolt.prepare(config, null, new OutputCollector(collector)); + bolt.setAsync(async); + bolt.setFireAndForget(fireAndForget); return bolt; } @@ -163,7 +217,6 @@ public class KafkaBoltTest { verify(collector).fail(tuple); } - private boolean verifyMessage(String key, String message) { long lastMessageOffset = KafkaUtils.getOffset(simpleConsumer, kafkaConfig.topic, 0, OffsetRequest.LatestTime()) - 1; ByteBufferMessageSet messageAndOffsets = KafkaUtils.fetchMessages(kafkaConfig, simpleConsumer, @@ -211,5 +264,4 @@ public class KafkaBoltTest { assertTrue(TupleUtils.isTick(tuple)); return tuple; } - }
