// Local contract model, not a Kafka or RabbitMQ client.
var kafka = new KafkaPartitionProgress();
var rabbit = new RabbitChannelProgress();
var first = new Delivery("order-1", 4, 17);
var second = new Delivery("order-2", 5, 18);
var handler = new Handler();
foreach (IProgress progress in new IProgress[] { kafka, rabbit })
{
    handler.Process(first, progress, fail: false);
    try { handler.Process(second, progress, fail: true); }
    catch (InvalidOperationException) { /* Retry the failed delivery before advancing. */ }
}
if (kafka.NextOffset != 5 || !rabbit.Acknowledged.SetEquals(new ulong[] { 17 }))
    throw new Exception("Failed work advanced progress");
handler.Process(second, kafka, fail: false);
handler.Process(second, rabbit, fail: false);
if (kafka.NextOffset != 6 || !rabbit.Acknowledged.Contains(18))
    throw new Exception("Successful retry did not advance progress");
Console.WriteLine("OK: failure unconfirmed; retry completed");

record Delivery(string Id, long Offset, ulong Tag);
interface IProgress { void Complete(Delivery delivery); }
sealed class KafkaPartitionProgress : IProgress
{
    // Exactly one partition, sequential processing, no rebalance in this model.
    public long NextOffset { get; private set; } = 4;
    public void Complete(Delivery delivery) => NextOffset = delivery.Offset + 1;
}
sealed class RabbitChannelProgress : IProgress
{
    // Exactly one channel; delivery tags are scoped to this channel.
    public HashSet<ulong> Acknowledged { get; } = new();
    public void Complete(Delivery delivery) => Acknowledged.Add(delivery.Tag);
}
sealed class Handler
{
    public void Process(Delivery delivery, IProgress progress, bool fail)
    {
        if (fail) throw new InvalidOperationException("Processing failed");
        // A real adapter must persist its business effect before confirming progress.
        progress.Complete(delivery);
    }
}