[
https://issues.apache.org/jira/browse/STORM-2675?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16113299#comment-16113299
]
Preet Puri commented on STORM-2675:
-----------------------------------
thanks Stig for analysis.. somehow I was thrown out by
https://github.com/apache/storm/blob/master/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java#L167
where Kafka consumer looks at last offset from Kafka __consumer_offset topic
and since we are not updating the offset in Kafka its always start from
earliest message in case of UNCOMMITTED_EARLIEST strategy?
So now my doubt is even if we fix the issue with Zookeeper metadata parsing
isn't the above logic going to ignore that information. To me it seems like
incomplete implementation?
> KafkaTridentSpoutOpaque not committing offsets to Kafka
> -------------------------------------------------------
>
> Key: STORM-2675
> URL: https://issues.apache.org/jira/browse/STORM-2675
> Project: Apache Storm
> Issue Type: Bug
> Components: storm-kafka-client
> Affects Versions: 1.1.0
> Reporter: Preet Puri
> Assignee: Stig Rohde Døssing
>
> Every time I restart the topology the spout was picking the earliest message
> even though poll strategy is set UNCOMMITTED_EARLIEST. I looked at Kafka's
> __consumer_offsets topic to see if spout (consumer) is committing the offsets
> but did not find any commits. I am not even able to locate the code in the
> KafkaTridentSpoutEmitter class where we are updating the commits?
> conf.put(Config.TOPOLOGY_DEBUG, true);
> conf.put(Config.TOPOLOGY_WORKERS, 1);
> conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 4); //tried with1 as well
> conf.put(Config.TRANSACTIONAL_ZOOKEEPER_ROOT, "/aggregate");
> conf.put(Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, Arrays.asList(new
> String[]{"localhost"}));
> conf.put(Config.TRANSACTIONAL_ZOOKEEPER_PORT, 2181);
> protected static KafkaSpoutConfig<String, String>
> getPMStatKafkaSpoutConfig() {
> ByTopicRecordTranslator<String, String> byTopic =
> new ByTopicRecordTranslator<>((r) -> new Values(r.topic(), r.key(),
> r.value()),
> new Fields(TOPIC, PARTITION_KEY, PAYLOAD), SENSOR_STREAM);
> return new KafkaSpoutConfig.Builder<String,
> String>(Utils.getBrokerHosts(),
> StringDeserializer.class, null, Utils.getKafkaEnrichedPMSTopicName())
> .setMaxPartitionFectchBytes(10 * 1024) // 10 KB
> .setRetry(getRetryService())
> .setOffsetCommitPeriodMs(10_000)
>
> .setFirstPollOffsetStrategy(FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST)
> .setMaxUncommittedOffsets(250)
> .setProp("value.deserializer",
> "io.confluent.kafka.serializers.KafkaAvroDeserializer")
> .setProp("schema.registry.url","http://localhost:8081";)
> .setProp("specific.avro.reader",true)
> .setGroupId(AGGREGATION_CONSUMER_GROUP)
> .setRecordTranslator(byTopic).build();
> }
> Stream pmStatStream =
> topology.newStream("statStream", new
> KafkaTridentSpoutOpaque<>(getPMStatKafkaSpoutConfig())).parallelismHint(1)
> storm-version - 1.1.0
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
