Error during SendOffsetsToTransaction : Operation not valid in state Ready

[LGouellec]

Description

Scenario :

• 1. I run a simple application which consume a topic (10 partitions) and display the message in the console output using EOS
• 2. I deploy the first instance, every thing is OK
• 3. I scale to 2, a rebalance occurs, 5 partitions are consumed for each instance
• 4. After some seconds, in the instance 1, my application crash because they can't commit the current transaction with a partition correctly assigned to this instance. The message is some weirds "Operation not valid in state Ready error".

partition.assigment.strategy = range

Offsets send in this transaction : input [[1]] @39736
dbug: Streamiz.Kafka.Net.Kafka.Internal.KafkaLoggerAdapter[0]
      stream-thread[sample-streamiz-demo-5dc30456-b375-4aef-ae00-943cad03edb0-stream-thread-0] Log producer sample-streamiz-demo-5dc30456-b375-4aef-ae00-943cad03edb0-stream-thread-0-0-1-streamiz-producer#producer-16 - [thrd:app]: Transactional API called: send_offsets_to_transaction (in txn state Ready, idemp state Assigned)
dbug: Streamiz.Kafka.Net.Kafka.Internal.KafkaLoggerAdapter[0]
      Log producer sample-streamiz-demo-5dc30456-b375-4aef-ae00-943cad03edb0-stream-thread-0-0-4-streamiz-producer#producer-19 - [thrd:main]: Querying for transaction coordinator: Coordinator query timer
dbug: Streamiz.Kafka.Net.Kafka.Internal.KafkaLoggerAdapter[0]
      Log producer sample-streamiz-demo-5dc30456-b375-4aef-ae00-943cad03edb0-stream-thread-0-0-4-streamiz-producer#producer-19 - [thrd:main]: localhost:9092/1: FindCoordinator response: Transaction coordinator is broker 1 (localhost:9092)
fail: Streamiz.Kafka.Net.Processors.StreamThread[0]
      stream-thread[sample-streamiz-demo-5dc30456-b375-4aef-ae00-943cad03edb0-stream-thread-0] Encountered the following unexpected Kafka exception during processing, this usually indicate Streams internal errors:
      Confluent.Kafka.KafkaException: Operation not valid in state Ready
         at Confluent.Kafka.Impl.SafeKafkaHandle.SendOffsetsToTransaction(IEnumerable`1 offsets, IConsumerGroupMetadata groupMetadata, Int32 millisecondsTimeout)
         at Confluent.Kafka.Producer`2.SendOffsetsToTransaction(IEnumerable`1 offsets, IConsumerGroupMetadata groupMetadata, TimeSpan timeout)

Checklist

Please provide the following information:

• [ ] A complete (i.e. we can run it), minimal program demonstrating the problem. No need to supply a project file.
• [X] Confluent.Kafka nuget version. (1.9.3)
• [X] Apache Kafka version. (3.1.0)
• [X] Client configuration. (default, just using EOS)
• [ ] Operating system.
• [X] Provide logs (with "debug" : "..." as necessary in configuration).
• [ ] Provide broker log excerpts.
• [ ] Critical issue.

[LGouellec]

If I switch the partition assigment strategy to CooperativeSticky, I have an other error during the same phase at the same method SendOffsetsToTransaction :

 Log producer sample-streamiz-demo-28aa5c7c-e134-4e76-8758-3298b0eff6f8-stream-thread-0-0-0-streamiz-producer#producer-5 - [thrd:main]: AddOffsetsToTxn response from TxnCoordinator/1: NO_ERROR ()
dbug: Streamiz.Kafka.Net.Kafka.Internal.KafkaLoggerAdapter[0]
      Log producer sample-streamiz-demo-28aa5c7c-e134-4e76-8758-3298b0eff6f8-stream-thread-0-0-0-streamiz-producer#producer-5 - [thrd:main]: Current transaction failed in state InTransaction: Failed to commit offsets to transaction on broker localhost:9092/1: input[0](Broker: Specified group generation id is not valid) (after 214ms) (ILLEGAL_GENERATION)
dbug: Streamiz.Kafka.Net.Kafka.Internal.KafkaLoggerAdapter[0]
      Log producer sample-streamiz-demo-28aa5c7c-e134-4e76-8758-3298b0eff6f8-stream-thread-0-0-0-streamiz-producer#producer-5 - [thrd:main]: Transaction state change InTransaction -> AbortableError
stream-thread[sample-streamiz-demo-28aa5c7c-e134-4e76-8758-3298b0eff6f8-stream-thread-0] Encountered the following unexpected Kafka exception during processing, this usually indicate Streams internal errors:
Confluent.Kafka.KafkaTxnRequiresAbortException: Failed to commit offsets to transaction on broker localhost:9092/1: input[0](Broker: Specified group generation id is not valid) (after 214ms)
         at Confluent.Kafka.Impl.SafeKafkaHandle.SendOffsetsToTransaction(IEnumerable`1 offsets, IConsumerGroupMetadata groupMetadata, Int32 millisecondsTimeout)