Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/1808#discussion_r91361471
--- Diff:
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java
---
@@ -60,123 +68,197 @@
* If no offset has been committed, it behaves as LATEST.</li>
* </ul>
* */
- public enum FirstPollOffsetStrategy {
+ public static enum FirstPollOffsetStrategy {
EARLIEST,
LATEST,
UNCOMMITTED_EARLIEST,
UNCOMMITTED_LATEST }
-
- // Kafka consumer configuration
- private final Map<String, Object> kafkaProps;
- private final Deserializer<K> keyDeserializer;
- private final Deserializer<V> valueDeserializer;
- private final long pollTimeoutMs;
-
- // Kafka spout configuration
- private final long offsetCommitPeriodMs;
- private final int maxRetries;
- private final int maxUncommittedOffsets;
- private final FirstPollOffsetStrategy firstPollOffsetStrategy;
- private final KafkaSpoutStreams kafkaSpoutStreams;
- private final KafkaSpoutTuplesBuilder<K, V> tuplesBuilder;
- private final KafkaSpoutRetryService retryService;
-
- private KafkaSpoutConfig(Builder<K,V> builder) {
- this.kafkaProps = setDefaultsAndGetKafkaProps(builder.kafkaProps);
- this.keyDeserializer = builder.keyDeserializer;
- this.valueDeserializer = builder.valueDeserializer;
- this.pollTimeoutMs = builder.pollTimeoutMs;
- this.offsetCommitPeriodMs = builder.offsetCommitPeriodMs;
- this.maxRetries = builder.maxRetries;
- this.firstPollOffsetStrategy = builder.firstPollOffsetStrategy;
- this.kafkaSpoutStreams = builder.kafkaSpoutStreams;
- this.maxUncommittedOffsets = builder.maxUncommittedOffsets;
- this.tuplesBuilder = builder.tuplesBuilder;
- this.retryService = builder.retryService;
+
+ public static Builder<String, String> builder(String bootstrapServers,
String ... topics) {
+ return new Builder<>(bootstrapServers, new StringDeserializer(),
new StringDeserializer(), topics);
}
-
- private Map<String, Object> setDefaultsAndGetKafkaProps(Map<String,
Object> kafkaProps) {
+
+ public static Builder<String, String> builder(String bootstrapServers,
Collection<String> topics) {
+ return new Builder<>(bootstrapServers, new StringDeserializer(),
new StringDeserializer(), topics);
--- End diff --
I don't think this works when the topology is serialized. See
https://github.com/apache/storm/pull/1696/files#diff-d2c28214a421176fceab415ffe4c95c0R24
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
