Fix Redpanda Migrator consumer group offset migration logic

[mihaitodor]

• Field is_high_watermark added to the redpanda_migrator_offsets output.
• Metadata field kafka_is_high_watermark added to the redpanda_migrator_offsets input.
• Fixed a bug with the redpanda_migrator_offsets input and output where the consumer group update migration logic based on timestamp lookup should no longer skip ahead in the destination cluster. This should enforce at-least-once delivery guarantees.
• The redpanda_migrator_bundle output no longer drops messages if either the redpanda_migrator or the redpanda_migrator_offsets child output throws an error. Connect will keep retrying to write the messages and apply backpressure to the input.

TODO:

• Push extra integration tests

[mihaitodor]

We really shouldn't be dropping messages here... I left this as a TODO initially thinking that we might want to configure DLQs, but it's best to just block if we can't write data. If it's a transitive issue, it will resume when the broker becomes responsive again. If not, an operator will have to look into it. Restarting Migrator should be a safe operation (even if it potentially introduces duplicate records).

[mihaitodor]

Same logic here, just repeated for the case when we don't use schemas.

[rockwotj]

Generally looks great! Is it possible to write any tests for this case? I will take a deeper look at the PR on Monday. Thanks Mihai

[mihaitodor]

Thank you for having a look Tyler! By the way, I think we won't have to decrement the timestamps in any way, since ListOffsetsAfterMilli() is inclusive as you mentioned in twmb/franz-go#897. Will think more through it as I clean up the tests...

[rockwotj]

I believe we can replace this function with a ListEndOffsets call right? It should be max(ListEndOffsets - 1, 0) I think because if the topic is empty you'll get back a high watermark that matches the log start position:

PARTITIONS
==========
PARTITION  LEADER  EPOCH  REPLICAS  LOG-START-OFFSET  HIGH-WATERMARK
0          0       1      [0]       0                 0

Also possible after retention kicks in

PARTITIONS
==========
PARTITION  LEADER  EPOCH  REPLICAS  LOG-START-OFFSET  HIGH-WATERMARK
0          0       1      [0]       1                 1

[mihaitodor]

I believe we can replace this function with a ListEndOffsets call right?

That's what I wanted to do... However, ListEndOffsets returns the offset after the offset of the last message (which I denoted as "last offset" - not sure what's a better name for it) in the topic. Here's an experiment:

> docker run --rm -it --name=source -p 8081:8081 -p 9092:9092 -p 9644:9644 redpandadata/redpanda redpanda start --node-id 0 --mode dev-container --set "rpk.additional_start_flags=[--reactor-backend=epoll]" --kafka-addr 0.0.0.0:9092 --advertise-kafka-addr host.docker.internal:9092 --schema-registry-addr 0.0.0.0:8081

$ rpk topic create test -p1 -X brokers=localhost:9092
TOPIC  STATUS
test   OK
$ echo foobar | rpk topic produce test -X brokers=localhost:9092
Produced to partition 0 at offset 0 with timestamp 1739834099974.

getLastRecordOffset():

Offset:  0

ListEndOffsets() / ListCommittedOffsets():

Offset_end:  1
Offset_comm:  1

Click this to reveal the messy code if you wish to play with it:

getLastRecordOffset():

func main() {
	seeds := []string{"localhost:9092"}
	consumeStart := map[string]map[int32]kgo.Offset{
		"test": {
			// The default offset begins at the end.
			0: kgo.NewOffset().Relative(-1),
		},
	}

	for {
		client, err := kgo.NewClient([]kgo.Opt{kgo.SeedBrokers(seeds...), kgo.ConsumePartitions(consumeStart)}...)
		if err != nil {
			log.Fatalf("failed to create Kafka client: %s", err)
		}

		ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
		fetches := client.PollFetches(ctx)
		if fetches.IsClientClosed() {
			log.Fatalf("failed to read message: client closed")
		}

		if err := fetches.Err(); err != nil {
			log.Fatalf("failed to read message: %s", err)
		}

		it := fetches.RecordIter()
		if it.Done() {
			log.Fatalf("couldn't find message: %s", err)
		}

		rec := it.Next()

		fmt.Println("Offset: ", rec.Offset)
		fmt.Println("Timestamp: ", rec.Timestamp)

		client.Close()

		time.Sleep(200 * time.Millisecond)
	}
}

ListEndOffsets() / ListCommittedOffsets():

func main() {
	seeds := []string{"localhost:9092"}

	client, err := kgo.NewClient([]kgo.Opt{kgo.SeedBrokers(seeds...)}...)
	if err != nil {
		log.Fatalf("failed to create Kafka client: %s", err)
	}
	defer client.Close()
	adm := kadm.NewClient(client)
	for {

		offsets, err := adm.ListEndOffsets(context.Background(), "test")
		if err != nil {
			log.Fatalf("failed to read offsets: %s", err)
		}

		o, ok := offsets.Lookup("test", 0)
		if !ok {
			log.Fatalf("failed to find offset for test-0")
		}

		fmt.Println("Offset_end: ", o.Offset)
		fmt.Println("Timestamp_end: ", o.Timestamp)

		offsets, err = adm.ListCommittedOffsets(context.Background(), "test")
		if err != nil {
			log.Fatalf("failed to read offsets: %s", err)
		}

		o, ok = offsets.Lookup("test", 0)
		if !ok {
			log.Fatalf("failed to find offset for test-0")
		}

		fmt.Println("Offset_comm: ", o.Offset)
		fmt.Println("Timestamp_comm: ", o.Timestamp)

		time.Sleep(200 * time.Millisecond)
	}
}

[rockwotj]

That's what I wanted to do... However, ListEndOffsets returns the offset after the offset of the last message (which I denoted as "last offset" - not sure what's a better name for it) in the topic. Here's an experiment:

Yeah I think we just do end offset (known as High Watermark or HWM) minus one.

[rockwotj]

Just one question about retries and then I think it LGTM

[rockwotj]

It's more about when we commit not what we read? So do we ack the messages after check returns true?

[mihaitodor]

So do we ack the messages after check returns true?

Yep, if I use read_until there, instead of reading 5 messages out of 10, it reads about 6-7... read_until should tell the child input to shut down at that point, but it's not immediate, so I think it gets a bit of time to ack before kafka_franz tears down the franz-go client. I'm not entirely sure how it's wired under the hood, but I recall bumping into issues with read_until before. For example, this generates 10 messages:

input:
  read_until:
    check: counter() == 2
    input:
      generate:
        count: 20
        batch_size: 5
        interval: 0s
        mapping: root = counter()

[rockwotj]

Yeah I was reading the code for read_until and was surprised that check only runs on the first message of every batch and the whole batch is still passed through..

[mihaitodor]

I did the cleanup & I'll merge it later tomorrow in case you think of other stuff I should cover in there. Thank you again!