基于confluent-dotnet的Kafka集成测试问题：竞态条件与重复分区分配

Hey there! Let’s tackle your Confluent .NET (librdkafka-backed) Kafka testing questions one by one—async Kafka code can be a real headache for deterministic tests, so I feel your pain.

1. Writing Race-Condition-Free Tests for "Consume From End + Produce One Message" Scenario #

The core issue here is that consumer initialization (offset positioning, partition assignment) happens asynchronously, so sending a message right after Subscribe/Assign leads to a race where the consumer hasn’t yet caught up to the partition end. Here’s a reliable approach:

Step-by-Step Solution #

• Wait for partition assignment to complete: Use a TaskCompletionSource to block until the consumer confirms it’s been assigned partitions.
• Explicitly seek to the partition end: Even if you set AutoOffsetReset = Latest, the consumer might not have finished loading the latest offset before you send your message. Manually seeking to the high watermark ensures you’re starting exactly at the end of the partition.
• Produce only after confirming the consumer is ready: Once you’ve completed the above steps, send your message and then consume with a reasonable timeout.

Example Test Code #

[Fact]
public async Task ConsumerReceivesOnlyNewMessageAfterStartingFromEnd()
{
    // Use a unique group ID to avoid cross-test contamination
    var consumerConfig = new ConsumerConfig
    {
        BootstrapServers = "localhost:9092",
        GroupId = Guid.NewGuid().ToString(),
        AutoOffsetReset = AutoOffsetReset.Latest,
        EnableAutoCommit = false, // Disable auto-commit for test determinism
        EnablePartitionEof = true
    };

    using var consumer = new ConsumerBuilder<Ignore, string>(consumerConfig).Build();
    var partitionAssignmentTcs = new TaskCompletionSource<bool>();
    var assignedPartitions = new List<TopicPartition>();

    // Subscribe with a callback to track partition assignment
    consumer.Subscribe("test-topic", (_, partitions) =>
    {
        assignedPartitions.AddRange(partitions);
        partitionAssignmentTcs.SetResult(true);
    });

    // Wait until partitions are assigned
    await partitionAssignmentTcs.Task;

    // Seek to the end of each assigned partition
    foreach (var tp in assignedPartitions)
    {
        var watermarkOffsets = consumer.GetWatermarkOffsets(tp);
        consumer.Seek(new TopicPartitionOffset(tp, watermarkOffsets.High));
    }

    // Produce the test message
    using var producer = new ProducerBuilder<Ignore, string>(new ProducerConfig { BootstrapServers = "localhost:9092" }).Build();
    const string testMessage = "test-message-123";
    await producer.ProduceAsync("test-topic", new Message<Ignore, string> { Value = testMessage });
    producer.Flush(TimeSpan.FromSeconds(5)); // Ensure message is written to Kafka

    // Consume and assert
    var consumeResult = consumer.Consume(TimeSpan.FromSeconds(5));
    Assert.NotNull(consumeResult);
    Assert.Equal(testMessage, consumeResult.Message.Value);

    // Verify no additional messages are consumed
    var eofResult = consumer.Consume(TimeSpan.FromSeconds(1));
    Assert.True(eofResult.IsPartitionEOF);
}

2. Is the Consumer Offset Determined After Calling Partition.Assign? #

Short answer: No, not immediately. When you call Assign (or Subscribe), the consumer only confirms it’s been assigned the specified partitions—the offset initialization (following AutoOffsetReset rules) happens in the background. The OnPartitionAssigned callback only provides TopicPartition because the offset hasn’t been resolved yet.

Why This Matters #

Even if you set AutoOffsetReset = Latest, the consumer might still be loading the latest offset from Kafka or local storage when you send your message. This is why manually calling Seek to the high watermark (as shown in the test above) is critical for deterministic tests—it forces the consumer to jump directly to the end of the partition.

If you want to verify the offset position after assignment, you can use consumer.Position(tp) once the consumer has had time to initialize, but relying on Seek is more reliable for testing.

3. Why Are Duplicate Partitions Assigned (More Than Exist)? #

In a healthy Kafka cluster (no broker failures), this is almost always a test code bug, not a librdkafka/Confluent .NET issue. Here are the most common causes:

• Manual Assign with duplicate partitions: If you’re using manual partition assignment (not consumer groups), double-check that you’re not adding the same TopicPartition multiple times to the assignment list. For example:

  // Bug: Duplicate partition 0 added
  var badPartitions = new List<TopicPartition>
  {
      new("test-topic", new Partition(0)),
      new("test-topic", new Partition(0))
  };
  consumer.Assign(badPartitions);
  

• Accidental repeated Assign calls: If your test code calls Assign multiple times (e.g., in an async callback that fires twice), the consumer will replace its assignment each time—but if you pass overlapping partitions, you might end up with duplicates.
• Incorrect partition count calculation: If you’re generating partitions dynamically (e.g., Enumerable.Range(0, 4) for a topic with 3 partitions), you’ll end up with an invalid partition number, which can sometimes manifest as unexpected assignment behavior.

Consumer groups will never assign the same partition to multiple consumers in the same group (Kafka’s rebalancing ensures this), so this issue is exclusive to manual assignment scenarios.

内容的提问来源于stack exchange，提问作者JJ15k