Loop kafka-spout trident (OpaqueTridentKafkaSpout)

[Miguel Ángel Fernández Fernández <mi...@gmail.com>]

Hello,

I have a deployment with several topologies and, from time to time, one of
them is stuck in an infinite loop. In this case, kafka-spout emits always
the same tuple (or batch of tuples).

Until now, i usually clean kafka directories, even storm directories. This
solution is quite drastic, because it results in data loss. But i cannot
find another solution.

Is there any problem with my OpaqueTridentKafkaSpout configuration?

Here is my code:

ZkHosts brokerHosts = new ZkHosts(zookeepers);

TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(brokerHosts,
KAFKATOPIC);

 kafkaConfig.scheme = new SchemeAsMultiScheme(new
OperationBigDecimalProtobuffScheme());

OpaqueTridentKafkaSpout kafkaSpout = new
OpaqueTridentKafkaSpout(kafkaConfig);

Any suggestions?

Thanks in advance

Re: Loop kafka-spout trident (OpaqueTridentKafkaSpout)

[Jeffery Maass <ma...@gmail.com>]

A reliable spout will receive notification whenever a message has
successfully traveled the topology or whenever a message has failed part of
the topology. The spout gets to decide what to do about that failure. I
don't know what the OpaqueTridentKafkaSpout does about failures, but, if it
is generic code - not written by you, that it merely retries. the batch of
tuples. Assuming the cause of the retries is a failed tuple - find out
where the failure is occurring ( worker logs ) , then change your bolt's
code to handle that failure.

Thank you for your time!

+++++++++++++++++++++
Jeff Maass <ma...@gmail.com>
linkedin.com/in/jeffmaass
stackoverflow.com/users/373418/maassql
+++++++++++++++++++++


On Tue, May 19, 2015 at 3:31 AM, Miguel Ángel Fernández Fernández <
miguelangelprogramacion@gmail.com> wrote:

> Hello,
>
> I have a deployment with several topologies and, from time to time, one of
> them is stuck in an infinite loop. In this case, kafka-spout emits always
> the same tuple (or batch of tuples).
>
> Until now, i usually clean kafka directories, even storm directories. This
> solution is quite drastic, because it results in data loss. But i cannot
> find another solution.
>
> Is there any problem with my OpaqueTridentKafkaSpout configuration?
>
> Here is my code:
>
> ZkHosts brokerHosts = new ZkHosts(zookeepers);
>
> TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(brokerHosts,
> KAFKATOPIC);
>
>  kafkaConfig.scheme = new SchemeAsMultiScheme(new
> OperationBigDecimalProtobuffScheme());
>
> OpaqueTridentKafkaSpout kafkaSpout = new
> OpaqueTridentKafkaSpout(kafkaConfig);
>
> Any suggestions?
>
> Thanks in advance
>