[jira] [Comment Edited] (STORM-2915) How could I to get the fail Number in Bolt When I use Kafka Spout

["Stig Rohde Døssing (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/STORM-2915?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16457132#comment-16457132 ] 

Stig Rohde Døssing edited comment on STORM-2915 at 4/27/18 9:57 PM:
--------------------------------------------------------------------

Could you explain your use case? The KafkaSpout as-is can't emit the failure count for the tuple, but it would be possible to add this feature by modifying the RecordTranslator interface. I'm just wondering why you need it?

Edit: Regarding the failure reason, that isn't really possible right now. See this thread http://mail-archives.apache.org/mod_mbox/storm-user/201804.mbox/%3CCA%2BGk7M3jZKFphyONuB3DjDR0-pJmAmOnphgt%3DFBRpT7zy21O%2BA%40mail.gmail.com%3E


was (Author: srdo):
Could you explain your use case? The KafkaSpout as-is can't emit the failure count for the tuple, but it would be possible to add this feature by modifying the RecordTranslator interface. I'm just wondering why you need it?

> How could I to get the fail Number   in Bolt When I use  Kafka Spout
> --------------------------------------------------------------------
>
>                 Key: STORM-2915
>                 URL: https://issues.apache.org/jira/browse/STORM-2915
>             Project: Apache Storm
>          Issue Type: New Feature
>          Components: storm-kafka-client
>    Affects Versions: 1.0.2, 1.1.0, 1.0.3, 1.0.4, 1.1.1, 1.0.5
>            Reporter: Gergo Hong
>            Priority: Minor
>
> I want to get fail num in bolt , how could  I  to get it? 
> if  fail it  retry, I see This 
> if (!isScheduled || retryService.isReady(msgId)) {
>  final String stream = tuple instanceof KafkaTuple ? ((KafkaTuple) tuple).getStream() : Utils.DEFAULT_STREAM_ID;
>  if (!isAtLeastOnceProcessing()) {
>  if (kafkaSpoutConfig.isTupleTrackingEnforced()) {
>  collector.emit(stream, tuple, msgId);
>  LOG.trace("Emitted tuple [{}] for record [{}] with msgId [{}]", tuple, record, msgId);
>  } else {
>  collector.emit(stream, tuple);
>  LOG.trace("Emitted tuple [{}] for record [{}]", tuple, record);
>  }
>  } else {
>  emitted.add(msgId);
>  offsetManagers.get(tp).addToEmitMsgs(msgId.offset());
>  if (isScheduled) { // Was scheduled for retry and re-emitted, so remove from schedule.
>  retryService.remove(msgId);
>  }
>  collector.emit(stream, tuple, msgId);
>  tupleListener.onEmit(tuple, msgId);
>  LOG.trace("Emitted tuple [{}] for record [{}] with msgId [{}]", tuple, record, msgId);
>  }
>  return true;
> }



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)