diff --git a/docs/storm-kafka-client.md b/docs/storm-kafka-client.md index 33b4dafe68d..ada861992a0 100644 --- a/docs/storm-kafka-client.md +++ b/docs/storm-kafka-client.md @@ -16,8 +16,8 @@ You need to provide implementations for the following 2 interfaces These interfaces have 2 methods defined: ```java - K getKeyFromTuple(Tuple/TridentTuple tuple); - V getMessageFromTuple(Tuple/TridentTuple tuple); +K getKeyFromTuple(Tuple/TridentTuple tuple); +V getMessageFromTuple(Tuple/TridentTuple tuple); ``` As the name suggests, these methods are called to map a tuple to a Kafka key and a Kafka message. If you just want one field @@ -53,10 +53,9 @@ These are also defined in `org.apache.kafka.clients.producer.ProducerConfig` ### Using wildcard kafka topic match You can do a wildcard topic match by adding the following config -``` - Config config = new Config(); - config.put("kafka.topic.wildcard.match",true); - +```java +Config config = new Config(); +config.put("kafka.topic.wildcard.match",true); ``` After this you can specify a wildcard topic for matching e.g. clickstream.*.log. This will match all streams matching clickstream.my.log, clickstream.cart.log etc @@ -67,65 +66,65 @@ After this you can specify a wildcard topic for matching e.g. clickstream.*.log. For the bolt : ```java - TopologyBuilder builder = new TopologyBuilder(); - - Fields fields = new Fields("key", "message"); - FixedBatchSpout spout = new FixedBatchSpout(fields, 4, - new Values("storm", "1"), - new Values("trident", "1"), - new Values("needs", "1"), - new Values("javadoc", "1") - ); - spout.setCycle(true); - builder.setSpout("spout", spout, 5); - //set producer properties. - Properties props = new Properties(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("acks", "1"); - props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); - props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); - - KafkaBolt bolt = new KafkaBolt() - .withProducerProperties(props) - .withTopicSelector(new DefaultTopicSelector("test")) - .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper()); - builder.setBolt("forwardToKafka", bolt, 8).shuffleGrouping("spout"); - - Config conf = new Config(); - - StormSubmitter.submitTopology("kafkaboltTest", conf, builder.createTopology()); +TopologyBuilder builder = new TopologyBuilder(); + +Fields fields = new Fields("key", "message"); +FixedBatchSpout spout = new FixedBatchSpout(fields, 4, + new Values("storm", "1"), + new Values("trident", "1"), + new Values("needs", "1"), + new Values("javadoc", "1") +); +spout.setCycle(true); +builder.setSpout("spout", spout, 5); +//set producer properties. +Properties props = new Properties(); +props.put("bootstrap.servers", "localhost:9092"); +props.put("acks", "1"); +props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); +props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + +KafkaBolt bolt = new KafkaBolt() + .withProducerProperties(props) + .withTopicSelector(new DefaultTopicSelector("test")) + .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper()); +builder.setBolt("forwardToKafka", bolt, 8).shuffleGrouping("spout"); + +Config conf = new Config(); + +StormSubmitter.submitTopology("kafkaboltTest", conf, builder.createTopology()); ``` For Trident: ```java - Fields fields = new Fields("word", "count"); - FixedBatchSpout spout = new FixedBatchSpout(fields, 4, - new Values("storm", "1"), - new Values("trident", "1"), - new Values("needs", "1"), - new Values("javadoc", "1") - ); - spout.setCycle(true); - - TridentTopology topology = new TridentTopology(); - Stream stream = topology.newStream("spout1", spout); - - //set producer properties. - Properties props = new Properties(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("acks", "1"); - props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); - props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); - - TridentKafkaStateFactory stateFactory = new TridentKafkaStateFactory() - .withProducerProperties(props) - .withKafkaTopicSelector(new DefaultTopicSelector("test")) - .withTridentTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("word", "count")); - stream.partitionPersist(stateFactory, fields, new TridentKafkaStateUpdater(), new Fields()); - - Config conf = new Config(); - StormSubmitter.submitTopology("kafkaTridentTest", conf, topology.build()); +Fields fields = new Fields("word", "count"); +FixedBatchSpout spout = new FixedBatchSpout(fields, 4, + new Values("storm", "1"), + new Values("trident", "1"), + new Values("needs", "1"), + new Values("javadoc", "1") +); +spout.setCycle(true); + +TridentTopology topology = new TridentTopology(); +Stream stream = topology.newStream("spout1", spout); + +//set producer properties. +Properties props = new Properties(); +props.put("bootstrap.servers", "localhost:9092"); +props.put("acks", "1"); +props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); +props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + +TridentKafkaStateFactory stateFactory = new TridentKafkaStateFactory() + .withProducerProperties(props) + .withKafkaTopicSelector(new DefaultTopicSelector("test")) + .withTridentTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("word", "count")); +stream.partitionPersist(stateFactory, fields, new TridentKafkaStateUpdater(), new Fields()); + +Config conf = new Config(); +StormSubmitter.submitTopology("kafkaTridentTest", conf, topology.build()); ``` ## Reading From kafka (Spouts) @@ -142,10 +141,8 @@ a spout. `topics` The topics the spout will consume can either be a `Collection` of specific topic names (1 or more) or a regular expression `Pattern`, which specifies that any topics that match that regular expression will be consumed. -In the case of the Constructors you may also need to specify a key deserializer and a value deserializer. This is to help guarantee type safety through the use -of Java generics. The defaults are `StringDeserializer`s and can be overwritten by calling `setKeyDeserializer` and/or `setValueDeserializer`. -If these are set to null the code will fall back to what is set in the kafka properties, but it is preferable to be explicit here, again to maintain -type safety with the generics. +If you are using the Builder Constructors instead of one of the `builder` methods, you will also need to specify a key deserializer and a value deserializer. This is to help guarantee type safety through the use +of Java generics. The deserializers can be specified via the consumer properties set with `setProp`. See the KafkaConsumer configuration documentation for details. There are a few key configs to pay attention to. @@ -161,11 +158,7 @@ for the first time. Allowed values include By default the "topic", "partition", "offset", "key", and "value" will be emitted to the "default" stream. If you want to output entries to different streams based on the topic, storm provides `ByTopicRecordTranslator`. See below for more examples on how to use these. -`setProp` can be used to set kafka properties that do not have a convenience method. - -`setGroupId` lets you set the id of the kafka consumer group property "group.id' - -`setSSLKeystore` and `setSSLTruststore` allow you to configure SSL authentication. +`setProp` and `setProps` can be used to set KafkaConsumer properties. The list of these properties can be found in the KafkaConsumer configuration documentation on the [Kafka website](http://kafka.apache.org/documentation.html#consumerconfigs). ### Usage Examples @@ -229,7 +222,7 @@ output topic it will throw an exception and not continue. In most cases the built in SimpleRecordTranslator and ByTopicRecordTranslator should cover your use case. If you do run into a situation where you need a custom one then this documentation will describe how to do this properly, and some of the less than obvious classes involved. -The point of apply is to take a ConsumerRecord and turn it into a `List` that can be emitted. What is not obvious is how to tell the spout to emit it to a +The point of `apply` is to take a ConsumerRecord and turn it into a `List` that can be emitted. What is not obvious is how to tell the spout to emit it to a specific stream. To do this you will need to return an instance of `org.apache.storm.kafka.spout.KafkaTuple`. This provides a method `routedTo` that will say which specific stream the tuple should go to. @@ -243,7 +236,7 @@ Will cause the tuple to be emitted on the "bar" stream. Be careful when writing custom record translators because just like in a storm spout it needs to be self consistent. The `streams` method should return a full set of streams that this translator will ever try to emit on. Additionally `getFieldsFor` should return a valid Fields object for each of those -streams. If you are doing this for Trident a value must be in the List returned by apply for every field in the Fields object for that stream, +streams. If you are doing this for Trident a value must be in the List returned by `apply` for every field in the Fields object for that stream, otherwise trident can throw exceptions. @@ -327,7 +320,7 @@ When selecting a kafka client version, you should ensure - # Kafka Spout Performance Tuning -The Kafka spout provides two internal parameters to control its performance. The parameters can be set using the [KafkaSpoutConfig] (https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java) methods [setOffsetCommitPeriodMs] (https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java#L189-L193) and [setMaxUncommittedOffsets] (https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java#L211-L217). +The Kafka spout provides two internal parameters to control its performance. The parameters can be set using the [KafkaSpoutConfig](https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java) methods [setOffsetCommitPeriodMs](https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java#L189-L193) and [setMaxUncommittedOffsets](https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java#L211-L217). * "offset.commit.period.ms" controls how often the spout commits to Kafka * "max.uncommitted.offsets" controls how many offsets can be pending commit before another poll can take place @@ -337,7 +330,7 @@ The [Kafka consumer config] (http://kafka.apache.org/documentation.html#consumer * “fetch.min.bytes” * “fetch.max.wait.ms” -* [Kafka Consumer] (http://kafka.apache.org/090/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html) instance poll timeout, which is specified for each Kafka spout using the [KafkaSpoutConfig] (https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java) method [setPollTimeoutMs] (https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java#L180-L184) +* [Kafka Consumer](http://kafka.apache.org/090/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html) instance poll timeout, which is specified for each Kafka spout using the [KafkaSpoutConfig](https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java) method [setPollTimeoutMs](https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java#L180-L184)
Depending on the structure of your Kafka cluster, distribution of the data, and availability of data to poll, these parameters will have to be configured appropriately. Please refer to the Kafka documentation on Kafka parameter tuning. @@ -356,6 +349,7 @@ Currently the Kafka spout has has the following default values, which have shown If reliability isn't important to you -- that is, you don't care about losing tuples in failure situations --, and want to remove the overhead of tuple tracking, then you can run a KafkaSpout with AutoCommitMode. To enable it, you need to: + * set Config.TOPOLOGY_ACKERS to 0; * enable *AutoCommitMode* in Kafka consumer configuration; diff --git a/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountNamedTopics.java b/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountNamedTopics.java index 9ab00f5c04f..886d15de03b 100644 --- a/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountNamedTopics.java +++ b/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountNamedTopics.java @@ -24,6 +24,7 @@ import java.util.Arrays; import java.util.List; import java.util.concurrent.TimeUnit; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.storm.Config; @@ -63,8 +64,8 @@ public List apply(ConsumerRecord record) { protected KafkaSpoutConfig newKafkaSpoutConfig() { return KafkaSpoutConfig.builder(KAFKA_LOCAL_BROKER, TOPIC_1, TOPIC_2) - .setGroupId("kafkaSpoutTestGroup_" + System.nanoTime()) - .setMaxPartitionFectchBytes(200) + .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup_" + System.nanoTime()) + .setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 200) .setRecordTranslator(JUST_VALUE_FUNC, new Fields("str")) .setRetry(newRetryService()) .setOffsetCommitPeriodMs(10_000) diff --git a/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountWildcardTopics.java b/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountWildcardTopics.java index 62ccef04aa3..0be31277d20 100644 --- a/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountWildcardTopics.java +++ b/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountWildcardTopics.java @@ -21,6 +21,7 @@ import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST; import java.util.regex.Pattern; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.storm.kafka.spout.KafkaSpoutConfig; import org.apache.storm.tuple.Fields; @@ -31,8 +32,8 @@ public class TridentKafkaClientWordCountWildcardTopics extends TridentKafkaClien protected KafkaSpoutConfig newKafkaSpoutConfig() { return KafkaSpoutConfig.builder("127.0.0.1:9092", TOPIC_WILDCARD_PATTERN) - .setGroupId("kafkaSpoutTestGroup") - .setMaxPartitionFectchBytes(200) + .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup") + .setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 200) .setRecordTranslator((r) -> new Values(r.value()), new Fields("str")) .setRetry(newRetryService()) .setOffsetCommitPeriodMs(10_000) diff --git a/external/storm-kafka-client/pom.xml b/external/storm-kafka-client/pom.xml index 843868fbc51..6db04a0432d 100644 --- a/external/storm-kafka-client/pom.xml +++ b/external/storm-kafka-client/pom.xml @@ -152,7 +152,7 @@ maven-checkstyle-plugin - 9 + 6 diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java index 43a6e0b83ec..6f09f5f67ab 100644 --- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java +++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java @@ -27,7 +27,6 @@ import java.util.regex.Pattern; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff.TimeInterval; import org.apache.storm.tuple.Fields; @@ -47,6 +46,7 @@ public class KafkaSpoutConfig implements Serializable { public static final int DEFAULT_MAX_UNCOMMITTED_OFFSETS = 10_000_000; // 2s public static final long DEFAULT_PARTITION_REFRESH_PERIOD_MS = 2_000; + public static final FirstPollOffsetStrategy DEFAULT_FIRST_POLL_OFFSET_STRATEGY = FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST; public static final KafkaSpoutRetryService DEFAULT_RETRY_SERVICE = new KafkaSpoutRetryExponentialBackoff(TimeInterval.seconds(0), TimeInterval.milliSeconds(2), DEFAULT_MAX_RETRIES, TimeInterval.seconds(10)); @@ -56,6 +56,37 @@ public class KafkaSpoutConfig implements Serializable { public static final KafkaSpoutRetryService UNIT_TEST_RETRY_SERVICE = new KafkaSpoutRetryExponentialBackoff(TimeInterval.seconds(0), TimeInterval.milliSeconds(0), DEFAULT_MAX_RETRIES, TimeInterval.milliSeconds(0)); + + // Kafka consumer configuration + private final Map kafkaProps; + private final Subscription subscription; + private final long pollTimeoutMs; + + // Kafka spout configuration + private final RecordTranslator translator; + private final long offsetCommitPeriodMs; + private final int maxUncommittedOffsets; + private final FirstPollOffsetStrategy firstPollOffsetStrategy; + private final KafkaSpoutRetryService retryService; + private final long partitionRefreshPeriodMs; + private final boolean emitNullTuples; + + /** + * Creates a new KafkaSpoutConfig using a Builder. + * @param builder The Builder to construct the KafkaSpoutConfig from + */ + public KafkaSpoutConfig(Builder builder) { + this.kafkaProps = setDefaultsAndGetKafkaProps(builder.kafkaProps); + this.subscription = builder.subscription; + this.translator = builder.translator; + this.pollTimeoutMs = builder.pollTimeoutMs; + this.offsetCommitPeriodMs = builder.offsetCommitPeriodMs; + this.firstPollOffsetStrategy = builder.firstPollOffsetStrategy; + this.maxUncommittedOffsets = builder.maxUncommittedOffsets; + this.retryService = builder.retryService; + this.partitionRefreshPeriodMs = builder.partitionRefreshPeriodMs; + this.emitNullTuples = builder.emitNullTuples; + } /** * The offset used by the Kafka spout in the first poll to Kafka broker. The choice of this parameter will affect the number of consumer @@ -77,163 +108,50 @@ public static enum FirstPollOffsetStrategy { EARLIEST, LATEST, UNCOMMITTED_EARLIEST, - UNCOMMITTED_LATEST } - - public static Builder builder(String bootstrapServers, String ... topics) { - return new Builder<>(bootstrapServers, StringDeserializer.class, StringDeserializer.class, topics); - } - - public static Builder builder(String bootstrapServers, Collection topics) { - return new Builder<>(bootstrapServers, StringDeserializer.class, StringDeserializer.class, topics); - } - - public static Builder builder(String bootstrapServers, Pattern topics) { - return new Builder<>(bootstrapServers, StringDeserializer.class, StringDeserializer.class, topics); - } - - private static Map setDefaultsAndGetKafkaProps(Map kafkaProps) { - // set defaults for properties not specified - if (!kafkaProps.containsKey(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)) { - kafkaProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); - } - return kafkaProps; + UNCOMMITTED_LATEST } public static class Builder { private final Map kafkaProps; private Subscription subscription; - private final SerializableDeserializer keyDes; - private final Class> keyDesClazz; - private final SerializableDeserializer valueDes; - private final Class> valueDesClazz; private RecordTranslator translator; private long pollTimeoutMs = DEFAULT_POLL_TIMEOUT_MS; private long offsetCommitPeriodMs = DEFAULT_OFFSET_COMMIT_PERIOD_MS; - private FirstPollOffsetStrategy firstPollOffsetStrategy = FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST; + private FirstPollOffsetStrategy firstPollOffsetStrategy = DEFAULT_FIRST_POLL_OFFSET_STRATEGY; private int maxUncommittedOffsets = DEFAULT_MAX_UNCOMMITTED_OFFSETS; private KafkaSpoutRetryService retryService = DEFAULT_RETRY_SERVICE; private long partitionRefreshPeriodMs = DEFAULT_PARTITION_REFRESH_PERIOD_MS; private boolean emitNullTuples = false; public Builder(String bootstrapServers, String ... topics) { - this(bootstrapServers, (SerializableDeserializer) null, (SerializableDeserializer) null, new NamedSubscription(topics)); - } - - public Builder(String bootstrapServers, SerializableDeserializer keyDes, SerializableDeserializer valDes, String ... topics) { - this(bootstrapServers, keyDes, valDes, new NamedSubscription(topics)); - } - - public Builder(String bootstrapServers, SerializableDeserializer keyDes, - SerializableDeserializer valDes, Collection topics) { - this(bootstrapServers, keyDes, valDes, new NamedSubscription(topics)); - } - - public Builder(String bootstrapServers, SerializableDeserializer keyDes, - SerializableDeserializer valDes, Pattern topics) { - this(bootstrapServers, keyDes, valDes, new PatternSubscription(topics)); - } - - public Builder(String bootstrapServers, SerializableDeserializer keyDes, - SerializableDeserializer valDes, Subscription subscription) { - this(bootstrapServers, keyDes, null, valDes, null, subscription); + this(bootstrapServers, new NamedSubscription(topics)); } - public Builder(String bootstrapServers, Class> keyDes, - Class> valDes, String ... topics) { - this(bootstrapServers, keyDes, valDes, new NamedSubscription(topics)); + public Builder(String bootstrapServers, Collection topics) { + this(bootstrapServers, new NamedSubscription(topics)); } - public Builder(String bootstrapServers, Class> keyDes, - Class> valDes, Collection topics) { - this(bootstrapServers, keyDes, valDes, new NamedSubscription(topics)); + public Builder(String bootstrapServers, Pattern topics) { + this(bootstrapServers, new PatternSubscription(topics)); } - public Builder(String bootstrapServers, Class> keyDes, - Class> valDes, Pattern topics) { - this(bootstrapServers, keyDes, valDes, new PatternSubscription(topics)); - } - - public Builder(String bootstrapServers, Class> keyDes, - Class> valDes, Subscription subscription) { - this(bootstrapServers, null, keyDes, null, valDes, subscription); - } - - private Builder(String bootstrapServers, SerializableDeserializer keyDes, - Class> keyDesClazz, - SerializableDeserializer valDes, Class> valDesClazz, Subscription subscription) { + /** + * Create a KafkaSpoutConfig builder with default property values and no key/value deserializers. + * @param bootstrapServers The bootstrap servers the consumer will use + * @param subscription The subscription defining which topics and partitions each spout instance will read. + */ + public Builder(String bootstrapServers, Subscription subscription) { kafkaProps = new HashMap<>(); if (bootstrapServers == null || bootstrapServers.isEmpty()) { throw new IllegalArgumentException("bootstrap servers cannot be null"); } kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); - this.keyDes = keyDes; - this.keyDesClazz = keyDesClazz; - this.valueDes = valDes; - this.valueDesClazz = valDesClazz; this.subscription = subscription; - this.translator = new DefaultRecordTranslator(); - } - - private Builder(Builder builder, SerializableDeserializer keyDes, Class> keyDesClazz, - SerializableDeserializer valueDes, Class> valueDesClazz) { - this.kafkaProps = new HashMap<>(builder.kafkaProps); - this.subscription = builder.subscription; - this.pollTimeoutMs = builder.pollTimeoutMs; - this.offsetCommitPeriodMs = builder.offsetCommitPeriodMs; - this.firstPollOffsetStrategy = builder.firstPollOffsetStrategy; - this.maxUncommittedOffsets = builder.maxUncommittedOffsets; - //this could result in a lot of class case exceptions at runtime, - // but because some translators will work no matter what the generics - // are I thought it best not to force someone to reset the translator - // when they change the key/value types. - this.translator = (RecordTranslator) builder.translator; - this.retryService = builder.retryService; - this.keyDes = keyDes; - this.keyDesClazz = keyDesClazz; - this.valueDes = valueDes; - this.valueDesClazz = valueDesClazz; - } - - /** - * Specifying this key deserializer overrides the property key.deserializer. If you have - * set a custom RecordTranslator before calling this it may result in class cast - * exceptions at runtime. - */ - public Builder setKey(SerializableDeserializer keyDeserializer) { - return new Builder<>(this, keyDeserializer, null, valueDes, valueDesClazz); + this.translator = new DefaultRecordTranslator<>(); } /** - * Specify a class that can be instantiated to create a key.deserializer - * This is the same as setting key.deserializer, but overrides it. If you have - * set a custom RecordTranslator before calling this it may result in class cast - * exceptions at runtime. - */ - public Builder setKey(Class> clazz) { - return new Builder<>(this, null, clazz, valueDes, valueDesClazz); - } - - /** - * Specifying this value deserializer overrides the property value.deserializer. If you have - * set a custom RecordTranslator before calling this it may result in class cast - * exceptions at runtime. - */ - public Builder setValue(SerializableDeserializer valueDeserializer) { - return new Builder<>(this, keyDes, keyDesClazz, valueDeserializer, null); - } - - /** - * Specify a class that can be instantiated to create a value.deserializer - * This is the same as setting value.deserializer, but overrides it. If you have - * set a custom RecordTranslator before calling this it may result in class cast - * exceptions at runtime. - */ - public Builder setValue(Class> clazz) { - return new Builder<>(this, keyDes, keyDesClazz, null, clazz); - } - - /** - * Set a Kafka property config. + * Set a {@link KafkaConsumer} property. */ public Builder setProp(String key, Object value) { kafkaProps.put(key, value); @@ -241,7 +159,7 @@ public Builder setProp(String key, Object value) { } /** - * Set multiple Kafka property configs. + * Set multiple {@link KafkaConsumer} properties. */ public Builder setProp(Map props) { kafkaProps.putAll(props); @@ -249,84 +167,18 @@ public Builder setProp(Map props) { } /** - * Set multiple Kafka property configs. + * Set multiple {@link KafkaConsumer} properties. */ public Builder setProp(Properties props) { - for (String name: props.stringPropertyNames()) { - kafkaProps.put(name, props.get(name)); - } + props.forEach((key, value) -> { + if (key instanceof String) { + kafkaProps.put((String)key, value); + } else { + throw new IllegalArgumentException("Kafka Consumer property keys must be Strings"); + } + }); return this; } - - /** - * Set the group.id for the consumers. - */ - public Builder setGroupId(String id) { - return setProp("group.id", id); - } - - /** - * reset the bootstrap servers for the Consumer. - */ - public Builder setBootstrapServers(String servers) { - return setProp(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers); - } - - /** - * The minimum amount of data the broker should return for a fetch request. - */ - public Builder setFetchMinBytes(int bytes) { - return setProp(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, bytes); - } - - /** - * The maximum amount of data per-partition the broker will return. - */ - public Builder setMaxPartitionFectchBytes(int bytes) { - return setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, bytes); - } - - /** - * The maximum number of records a poll will return. - */ - public Builder setMaxPollRecords(int records) { - return setProp(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, records); - } - - //Security Related Configs - - /** - * Configure the SSL Keystore for mutual authentication. - */ - public Builder setSSLKeystore(String location, String password) { - return setProp("ssl.keystore.location", location) - .setProp("ssl.keystore.password", password); - } - - /** - * Configure the SSL Keystore for mutual authentication. - */ - public Builder setSSLKeystore(String location, String password, String keyPassword) { - return setProp("ssl.key.password", keyPassword) - .setSSLKeystore(location, password); - } - - /** - * Configure the SSL Truststore to authenticate with the brokers. - */ - public Builder setSSLTruststore(String location, String password) { - return setSecurityProtocol("SSL") - .setProp("ssl.truststore.location", location) - .setProp("ssl.truststore.password", password); - } - - /** - * Protocol used to communicate with brokers. - * Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL. - */ - public Builder setSecurityProtocol(String protocol) { - return setProp("security.protocol", protocol); - } //Spout Settings /** @@ -435,82 +287,58 @@ public KafkaSpoutConfig build() { return new KafkaSpoutConfig<>(this); } } - - // Kafka consumer configuration - private final Map kafkaProps; - private final Subscription subscription; - private final SerializableDeserializer keyDes; - private final Class> keyDesClazz; - private final SerializableDeserializer valueDes; - private final Class> valueDesClazz; - private final long pollTimeoutMs; - - // Kafka spout configuration - private final RecordTranslator translator; - private final long offsetCommitPeriodMs; - private final int maxUncommittedOffsets; - private final FirstPollOffsetStrategy firstPollOffsetStrategy; - private final KafkaSpoutRetryService retryService; - private final long partitionRefreshPeriodMs; - private final boolean emitNullTuples; - + + /** - * Creates a new KafkaSpoutConfig using a Builder. - * @param builder The Builder to construct the KafkaSpoutConfig from + * Factory method that creates a Builder with String key/value deserializers. + * @param bootstrapServers The bootstrap servers for the consumer + * @param topics The topics to subscribe to + * @return The new builder */ - public KafkaSpoutConfig(Builder builder) { - this.kafkaProps = setDefaultsAndGetKafkaProps(builder.kafkaProps); - this.subscription = builder.subscription; - this.translator = builder.translator; - this.pollTimeoutMs = builder.pollTimeoutMs; - this.offsetCommitPeriodMs = builder.offsetCommitPeriodMs; - this.firstPollOffsetStrategy = builder.firstPollOffsetStrategy; - this.maxUncommittedOffsets = builder.maxUncommittedOffsets; - this.retryService = builder.retryService; - this.keyDes = builder.keyDes; - this.keyDesClazz = builder.keyDesClazz; - this.valueDes = builder.valueDes; - this.valueDesClazz = builder.valueDesClazz; - this.partitionRefreshPeriodMs = builder.partitionRefreshPeriodMs; - this.emitNullTuples = builder.emitNullTuples; + public static Builder builder(String bootstrapServers, String ... topics) { + return setStringDeserializers(new Builder<>(bootstrapServers, topics)); } - + /** - * Gets the properties that will be passed to the KafkaConsumer. - * @return The Kafka properties map + * Factory method that creates a Builder with String key/value deserializers. + * @param bootstrapServers The bootstrap servers for the consumer + * @param topics The topics to subscribe to + * @return The new builder */ - public Map getKafkaProps() { - return kafkaProps; + public static Builder builder(String bootstrapServers, Collection topics) { + return setStringDeserializers(new Builder<>(bootstrapServers, topics)); } - + /** - * Gets the Kafka record key deserializer. - * @return The key deserializer + * Factory method that creates a Builder with String key/value deserializers. + * @param bootstrapServers The bootstrap servers for the consumer + * @param topics The topic pattern to subscribe to + * @return The new builder */ - public Deserializer getKeyDeserializer() { - if (keyDesClazz != null) { - try { - return keyDesClazz.newInstance(); - } catch (InstantiationException | IllegalAccessException e) { - throw new RuntimeException("Could not instantiate key deserializer " + keyDesClazz); - } + public static Builder builder(String bootstrapServers, Pattern topics) { + return setStringDeserializers(new Builder<>(bootstrapServers, topics)); + } + + private static Builder setStringDeserializers(Builder builder) { + builder.setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + builder.setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + return builder; + } + + private static Map setDefaultsAndGetKafkaProps(Map kafkaProps) { + // set defaults for properties not specified + if (!kafkaProps.containsKey(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)) { + kafkaProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); } - return keyDes; + return kafkaProps; } /** - * Gets the Kafka record value deserializer. - * @return The value deserializer + * Gets the properties that will be passed to the KafkaConsumer. + * @return The Kafka properties map */ - public Deserializer getValueDeserializer() { - if (valueDesClazz != null) { - try { - return valueDesClazz.newInstance(); - } catch (InstantiationException | IllegalAccessException e) { - throw new RuntimeException("Could not instantiate value deserializer " + valueDesClazz); - } - } - return valueDes; + public Map getKafkaProps() { + return kafkaProps; } public Subscription getSubscription() { @@ -562,8 +390,6 @@ public boolean isEmitNullTuples() { public String toString() { return "KafkaSpoutConfig{" + "kafkaProps=" + kafkaProps - + ", key=" + getKeyDeserializer() - + ", value=" + getValueDeserializer() + ", pollTimeoutMs=" + pollTimeoutMs + ", offsetCommitPeriodMs=" + offsetCommitPeriodMs + ", maxUncommittedOffsets=" + maxUncommittedOffsets diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/SerializableDeserializer.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/SerializableDeserializer.java deleted file mode 100644 index 120260e75e7..00000000000 --- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/SerializableDeserializer.java +++ /dev/null @@ -1,26 +0,0 @@ -/* - * Copyright 2016 The Apache Software Foundation. - * - * 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. - */ - -package org.apache.storm.kafka.spout; - -import java.io.Serializable; -import org.apache.kafka.common.serialization.Deserializer; - -/** - * @param The type this deserializer deserializes to. - */ -public interface SerializableDeserializer extends Deserializer, Serializable { -} diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/KafkaConsumerFactoryDefault.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/KafkaConsumerFactoryDefault.java index c2838153896..9a8142a7990 100644 --- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/KafkaConsumerFactoryDefault.java +++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/KafkaConsumerFactoryDefault.java @@ -23,8 +23,7 @@ public class KafkaConsumerFactoryDefault implements KafkaConsumerFactory createConsumer(KafkaSpoutConfig kafkaSpoutConfig) { - return new KafkaConsumer<>(kafkaSpoutConfig.getKafkaProps(), - kafkaSpoutConfig.getKeyDeserializer(), kafkaSpoutConfig.getValueDeserializer()); + return new KafkaConsumer<>(kafkaSpoutConfig.getKafkaProps()); } } diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutManager.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutManager.java index cf790b27cfd..30d52cb219f 100644 --- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutManager.java +++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutManager.java @@ -49,8 +49,7 @@ public KafkaTridentSpoutManager(KafkaSpoutConfig kafkaSpoutConfig) { } KafkaConsumer createAndSubscribeKafkaConsumer(TopologyContext context) { - kafkaConsumer = new KafkaConsumer<>(kafkaSpoutConfig.getKafkaProps(), - kafkaSpoutConfig.getKeyDeserializer(), kafkaSpoutConfig.getValueDeserializer()); + kafkaConsumer = new KafkaConsumer<>(kafkaSpoutConfig.getKafkaProps()); kafkaSpoutConfig.getSubscription().subscribe(kafkaConsumer, new KafkaSpoutConsumerRebalanceListener(), context); return kafkaConsumer; diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java index e904e4c32d3..513db90cb45 100644 --- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java @@ -22,6 +22,8 @@ import static org.junit.Assert.assertTrue; import java.util.HashMap; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy; import org.junit.Test; @@ -34,8 +36,10 @@ public void testBasic() { assertNull(conf.getConsumerGroupId()); assertTrue(conf.getTranslator() instanceof DefaultRecordTranslator); HashMap expected = new HashMap<>(); - expected.put("bootstrap.servers", "localhost:1234"); - expected.put("enable.auto.commit", "false"); + expected.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:1234"); + expected.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); + expected.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + expected.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); assertEquals(expected, conf.getKafkaProps()); } diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/MaxUncommittedOffsetTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/MaxUncommittedOffsetTest.java index 0a6a16ba861..9ebdcf7fefb 100755 --- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/MaxUncommittedOffsetTest.java +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/MaxUncommittedOffsetTest.java @@ -32,6 +32,7 @@ import java.util.List; import java.util.Map; import java.util.stream.Collectors; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.storm.kafka.KafkaUnitRule; import org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration; @@ -59,7 +60,7 @@ public class MaxUncommittedOffsetTest { private final int initialRetryDelaySecs = 60; private final KafkaSpoutConfig spoutConfig = getKafkaSpoutConfigBuilder(kafkaUnitRule.getKafkaUnit().getKafkaPort()) .setOffsetCommitPeriodMs(commitOffsetPeriodMs) - .setMaxPollRecords(maxPollRecords) + .setProp(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords) .setMaxUncommittedOffsets(maxUncommittedOffsets) .setRetry(new KafkaSpoutRetryExponentialBackoff(KafkaSpoutRetryExponentialBackoff.TimeInterval.seconds(initialRetryDelaySecs), KafkaSpoutRetryExponentialBackoff.TimeInterval.seconds(0), 1, KafkaSpoutRetryExponentialBackoff.TimeInterval.seconds(initialRetryDelaySecs))) //Retry once after a minute diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/SingleTopicKafkaSpoutConfiguration.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/SingleTopicKafkaSpoutConfiguration.java index 48413becb40..62dbfe5122c 100644 --- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/SingleTopicKafkaSpoutConfiguration.java +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/SingleTopicKafkaSpoutConfiguration.java @@ -19,6 +19,7 @@ import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.storm.Config; import org.apache.storm.generated.StormTopology; import org.apache.storm.kafka.spout.KafkaSpout; @@ -50,8 +51,8 @@ public static KafkaSpoutConfig.Builder getKafkaSpoutConfigBuilder return KafkaSpoutConfig.builder("127.0.0.1:" + port, TOPIC) .setRecordTranslator((r) -> new Values(r.topic(), r.key(), r.value()), new Fields("topic", "key", "value"), STREAM) - .setGroupId("kafkaSpoutTestGroup") - .setMaxPollRecords(5) + .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup") + .setProp(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5) .setRetry(getRetryService()) .setOffsetCommitPeriodMs(10_000) .setFirstPollOffsetStrategy(EARLIEST) diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainNamedTopics.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainNamedTopics.java index 9459d4bd38c..50991e88685 100644 --- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainNamedTopics.java +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainNamedTopics.java @@ -23,6 +23,7 @@ import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.storm.Config; import org.apache.storm.LocalCluster; import org.apache.storm.StormSubmitter; @@ -98,7 +99,7 @@ protected KafkaSpoutConfig getKafkaSpoutConfig() { (r) -> new Values(r.topic(), r.partition(), r.offset(), r.key(), r.value()), new Fields("topic", "partition", "offset", "key", "value"), TOPIC_2_STREAM); return KafkaSpoutConfig.builder("127.0.0.1:9092", TOPICS) - .setGroupId("kafkaSpoutTestGroup") + .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup") .setRetry(getRetryService()) .setRecordTranslator(trans) .setOffsetCommitPeriodMs(10_000) diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainWildcardTopics.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainWildcardTopics.java index 5ea9d99f181..d50e2ab7516 100644 --- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainWildcardTopics.java +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainWildcardTopics.java @@ -21,6 +21,7 @@ import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST; import java.util.regex.Pattern; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.storm.generated.StormTopology; import org.apache.storm.kafka.spout.KafkaSpout; import org.apache.storm.kafka.spout.KafkaSpoutConfig; @@ -45,7 +46,7 @@ protected StormTopology getTopologyKafkaSpout() { protected KafkaSpoutConfig getKafkaSpoutConfig() { return KafkaSpoutConfig.builder("127.0.0.1:9092", TOPIC_WILDCARD_PATTERN) - .setGroupId("kafkaSpoutTestGroup") + .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup") .setRetry(getRetryService()) .setRecordTranslator((r) -> new Values(r.topic(), r.partition(), r.offset(), r.key(), r.value()), new Fields("topic", "partition", "offset", "key", "value"), STREAM)