System.ArgumentException: An item with the same key has already been added.

[edcastanohunter]

Hi here, I am having something strange regarding an error triggered in a stream that I am using. I got this error

System.ArgumentException: An item with the same key has already been added. Key: xy-nominal-tech-consumer-xyz-nominal-converter-x1-flow-converterx1conversion-f3f1a844-4370-4109-998e-af873a2ca889-stream-t at System.Collections.Generic.Dictionary`2.TryInsert(TKey key, TValue value, InsertionBehavior behavior)
   at System.Collections.Generic.Dictionary`2.Add(TKey key, TValue value)
   at Streamiz.Kafka.Net.Kafka.Internal.RecordCollector.Init(IProducer`2& producer)
   at Streamiz.Kafka.Net.Processors.StreamTask..ctor(String threadId, TaskId id, IEnumerable`1 partitions, ProcessorTopology processorTopology, IConsumer`2 consumer, IStreamConfig configuration, IKafkaSupplier kafkaSupplier, IProducer`2 producer, IChangelogRegister changelogRegister, StreamMetricsRegistry streamMetricsRegistry)
   at Streamiz.Kafka.Net.Processors.Internal.TaskCreator.CreateTask(IConsumer`2 consumer, TaskId id, IEnumerable`1 partitions)
   at Streamiz.Kafka.Net.Processors.Internal.AbstractTaskCreator`1.CreateTasks(IConsumer`2 consumer, IDictionary`2 tasksToBeCreated)
   at Streamiz.Kafka.Net.Processors.Internal.TaskManager.CreateTasks(ICollection`1 assignment)
   at Streamiz.Kafka.Net.Kafka.Internal.StreamsRebalanceListener.PartitionsAssigned(IConsumer`2 consumer, List`1 partitions)

I not sure it that might be for the length of the string. Could you help me figuring out?

[LGouellec]

Hi @edcastanohunter,

Which version are you using ? Can you share your configuration ?
FYI, I currently refactor this part of the code, will be release in 1.7.0.

Best regards,

[edcastanohunter]

I am using this libraries:

<Project Sdk="Microsoft.NET.Sdk">
    <PropertyGroup>
        <TargetFramework>net8.0</TargetFramework>
    </PropertyGroup>
...

        <PackageReference Include="Streamiz.Kafka.Net" Version="1.5.1" />
        <PackageReference Include="Confluent.Kafka" Version="2.3.0" />
        <PackageReference Include="Confluent.Kafka.Extensions" Version="1.0.0" />

[LGouellec]

Can you share your configuration as well ?

[edcastanohunter]

Sure:

"Kafka": {
    "ConsumerGroupPrefix": "prefix-",
    "KafkaTopic": "theTopic",
    "KafkaMessageValidMinutes": 30,
    "KafkaStreamId": "theStream",
    "NumStreamThreads": 4,
    "StreamConfig": {
      "SecurityProtocol": "SSL",
      "BootstrapServers": "theserverurl1:9092"
    }
  }

[LGouellec]

@edcastanohunter

Oh I see, as I said, this part of the code will be remove in 1.7.0.
Is it happens every time ? Can you try to upgrade in 1.6.0 please ?

Kr,

[edcastanohunter]

Hi @LGouellec, I have upgrade the library to 1.6, but we are still receiving the error

An item with the same key has already been added. Key: xy-service-dev2-provider-qrs-classify-xyz-conversion-flow-interpreterv2conversion-15975324-1b24-3d53-9642-da31e2da1425-strea

   at System.Collections.Generic.Dictionary`2.TryInsert(TKey key, TValue value, InsertionBehavior behavior)
   at System.Collections.Generic.Dictionary`2.Add(TKey key, TValue value)
   at Streamiz.Kafka.Net.Kafka.Internal.RecordCollector.Init(IProducer`2& producer)
   at Streamiz.Kafka.Net.Processors.StreamTask..ctor(String threadId, TaskId id, IEnumerable`1 partitions, ProcessorTopology processorTopology, IConsumer`2 consumer, IStreamConfig configuration, IKafkaSupplier kafkaSupplier, IProducer`2 producer, IChangelogRegister changelogRegister, StreamMetricsRegistry streamMetricsRegistry)
   at Streamiz.Kafka.Net.Processors.Internal.TaskCreator.CreateTask(IConsumer`2 consumer, TaskId id, IEnumerable`1 partitions)
   at Streamiz.Kafka.Net.Processors.Internal.AbstractTaskCreator`1.CreateTasks(IConsumer`2 consumer, IDictionary`2 tasksToBeCreated)
   at Streamiz.Kafka.Net.Processors.Internal.TaskManager.CreateTasks(ICollection`1 assignment)
   at Streamiz.Kafka.Net.Kafka.Internal.StreamsRebalanceListener.PartitionsAssigned(IConsumer`2 consumer, List`1 partitions)

Not sure if you need something else that help us identify this strange behavior. If you point me to the code is handling this I might be able to help.

[LGouellec]

Are you able to test with the last commit in the develop branch ?