Skip to main content
plapla
plapla
Silver ⭐️⭐️⭐️
January 23, 2024
Question

Nifi consume duplicate message problem from Kafka

Nifi version is 1.23.2 and kafka version 2.6.I use ConsumeKafka_1_0 1.23.2 to read messages from kafka.DetectDuplicate 1.23.2 processor is available, which deduplicates stream files on the same stream.However, stream read the same recording twice  and DetectDuplicate processor could not deduplicate the same message.I understand that it reads the same message because the offset  id value is same in two message.What could be the reason why the ConsumeKafka_1_0  processor reads the same message and DetectDuplicate  processor cannot detect same message?

 

plapla_0-1705994944172.png

 

plapla_1-1705995855058.png

plapla_2-1705995906341.png

 

 

10 replies

MattWho
Titanium ⭐️⭐️⭐️
MattWho
Titanium ⭐️⭐️⭐️
January 23, 2024

@plapla 

Couple things here:

1. You should be using the ConsumeKafka processor that matches your Kafka server version.  If your KafKa server is 2.6, you should be using the ConsumeKafka_2_6 processor.

2. In your detectDuplicate processor you are using ${key}.  How is this Attribute beig created on each FlowFile and where is it's value derived from?

Thanks,
Matt

Translate
plapla
plaplaAuthor
Silver ⭐️⭐️⭐️
January 24, 2024

Thanks for your return @MattWho 

1.I confirmed the kafka version,kafka version is 2.5.In this case, which version of the processor should I use?

2.I transform the content to the key attribute with the

EvaluateJsonPath processor.Then I use the key value in the detectDuplicate processor as I share below.I confirmed that the attribute of the flowfile that cannot be deduplicated is.Could this be related to the flowfile arriving at the same time?
 
plapla_0-1706080906260.pngplapla_1-1706080942033.png

 

Translate
plapla
plaplaAuthor
Silver ⭐️⭐️⭐️
January 26, 2024

Hi @MattWho ,

Do you have any comments on the subject?

Translate
    plapla
    plaplaAuthor
    Silver ⭐️⭐️⭐️
    January 30, 2024

    Hi @MattWho ,

    I updated the processor to ConsumeKafka_2_0.

    1.My Nifi is not clustered.It consists of a single node.

    2.distributedMapCacheClient and DistributedMapCacheServer as I share below;

     

    plapla_0-1706624669008.png

     

    plapla_1-1706624707586.png

    3.ConsumeKafka processor reads approximately 200 flowFiles in 5 minutes

    I configured the DetectDuplicate processor with an age off of 10 days.It doesn't matter if the processor can't deduplicate messages.It's bad for the ConsumeKafka to read the same message twice.

    Finally,I updated 10000 to 200 value on max poll records attributes ConsumeKafka.

     

     

     

    Translate
    MattWho
    Titanium ⭐️⭐️⭐️
    MattWho
    Titanium ⭐️⭐️⭐️
    January 31, 2024

    @plapla 
    The consumeKafka processor should not be reading the same message twice.  The processor should be maintaining local state (since you are not clustered) in NiFi's local state directory. Make sure that you are not having disk space or permissions issues that may prevent the writing of that local state.  You can write click on the ConsumeKafka processor to view the current stored state.

    The consumeKafka processor creates a consumer group using the GroupID configured in the processor, so make sure you do not have multiple consumeKafka processor consuming from the same Kafka Topic using the same Group ID.  For optimal performance the number of concurrent tasks configured on the consumeKafka processor should match the number fo partitions on target topic.

    Do you see any Kafka rebalance going on?   Will happen when you have more consumers than partitions in a the consumer group that is consuming from that topic.  A rebalance can affect the commit of the offset resulting in possible data duplication.

    Thanks,

    Matt

    Translate
      plapla
      plaplaAuthor
      Silver ⭐️⭐️⭐️
      February 14, 2024

      @MattWho 

      The reason for this situation is not because the Consume processor reads the same message twice or because the detect duplicate processor cannot capture the same message.It is caused by the putElasticsearchHttp processor adding data to the elastic cluster  twice.The putElasticsearchHttp  processor has received a timeout error and the flowFile goes to the failure transition.When the putElasticsearchHttp   processor receives a successful response, flowFile is deleted, but it is recorded twice in the elastic. environment.What could be the reason for this?

      Translate
        plapla
        plaplaAuthor
        Silver ⭐️⭐️⭐️
        February 15, 2024

        @MattWho 

        putElasticsearchHttp processor  connection Timeout parameter is 5 secs.Response timeout parameter is 15 secs.Would it be appropriate if I doubled these values?

        Translate
          plapla
          plaplaAuthor
          Silver ⭐️⭐️⭐️
          February 15, 2024

          @MattWho 

          I forgot to answer the question.If flowfile drop on failure and retry transition,I rotates flowFile same putElasticsearchHttp  processor.

           

          plapla_0-1707995978972.png

           

          Translate
            Terms & ConditionsAccessibility statement