Refactor Redpanda Migrator components #3026

mihaitodor · 2024-11-21T02:27:15Z

I hijacked this PR to address several issues:

Added

New redpanda_migrator_offsets input.
Fields offset_topic, offset_group, offset_partition, offset_commit_timestamp and offset_metadata added to the redpanda_migrator_offsets output.
Fields kafka_key and max_in_flight for the redpanda_migrator_offsets output are now deprecated.
Fields batching for the redpanda_migrator output is now deprecated.
Field topic_lag_refresh_period added to the redpanda and redpanda_common inputs.
Metric redpanda_lag now emitted by the redpanda and redpanda_common inputs.
Metadata kafka_lag now emitted by the redpanda and redpanda_common inputs.

Fixed

The redpanda_migrator_bundle output now skips schema ID translation when translate_schema_ids: false and schema_registry is configured.
The redpanda_migrator output no longer rejects messages if it can't perform schema ID translation.
The redpanda_migrator input no longer converts the kafka key to string.

Changed

The kafka_key and max_in_flight fields of the redpanda_migrator_offsets output are now deprecated.
Fields batch_size and multi_header for the redpanda_migrator input are now deprecated.
- The redpanda_migrator_bundle input and output now set labels for their subcomponents.
The redpanda_migrator input no longer emits tombstone messages.

Redpanda Migrator offset metadata

One quick way to test this is via the following config. Note how I overwrite kafka_offset_metadata to foobar in a mapping processor.

input:
  redpanda_migrator_bundle:
    redpanda_migrator:
      seed_brokers: [ "localhost:9092" ]
      topics:
        - '^[^_]' # Skip internal topics which start with `_`
      regexp_topics: true
      consumer_group: migrator_bundle
      start_from_oldest: true
      replication_factor_override: true
      replication_factor: -1

    schema_registry:
      url: http://localhost:8081
      include_deleted: true
      subject_filter: ""

output:
  processors:
    - switch:
        - check: metadata("input_label") == "redpanda_migrator_offsets_input"
          processors:
            - mapping: |
                meta kafka_offset_metadata = "foobar"
  redpanda_migrator_bundle:
    redpanda_migrator:
      seed_brokers: [ "localhost:9093" ]
      max_in_flight: 1
      replication_factor_override: true
      replication_factor: -1

    schema_registry:
      url: http://localhost:8082

mihaitodor · 2024-12-16T11:39:41Z

internal/impl/kafka/franz_reader_ordered.go

+		log:               res.Logger(),
+		shutSig:           shutdown.NewSignaller(),
+		clientOpts:        optsFn,
+		topicLagGauge:     res.Metrics().NewGauge("redpanda_lag", "topic", "partition"),


When I added the redpanda_migrator input, I had both this gauge and the kafka_lag metadata field. I don't know if we want any of these available by default. Also, should this gauge name be somehow derived from the actual input type (redpanda, redpanda_common, redpanda_migrator, redpanda_migrator_offsets)? It does get the label of the input if set, so maybe that's sufficient.

I think the label is enough. Do we really want this lag metric for all these inputs? Probably I would assume...

I also think it's a bit overkill and I don't recall now which conversation led to this pattern. I also emit the kafka_lag metadata field with each message, so one could add a metric processor in the pipeline which creates a gauge for topics as needed. One downside with this approach is if messages stop flowing completely, then this gauge wouldn't get any updates. I think the main idea was to make it easier for people to discover this metric, but it's not clear what the perf impact might be if we consume from thousands of topics, each having multiple partitions. Should I remove it? (cc @Jeffail)

I like having the metric emitted here, it's relatively cheap, and extracting from meta is awkward enough no one is going to do it willingly.

mihaitodor · 2024-12-31T15:10:07Z

internal/impl/kafka/franz_reader_ordered.go

@@ -76,18 +105,25 @@ type FranzReaderOrdered struct {

 // NewFranzReaderOrderedFromConfig attempts to instantiate a new
 // FranzReaderOrdered reader from a parsed config.
-func NewFranzReaderOrderedFromConfig(conf *service.ParsedConfig, res *service.Resources, optsFn func() ([]kgo.Opt, error)) (*FranzReaderOrdered, error) {
+func NewFranzReaderOrderedFromConfig(conf *service.ParsedConfig, res *service.Resources, clientOptsFn clientOptsFn, recordToMessageFn recordToMessageFn, preflightHookFn preflightHookFn, closeHookFn closeHookFn) (*FranzReaderOrdered, error) {


This constructor got a bit messy to use... It can be hard to tell which of these funcs is set to nil at the call site and one can easily mix them up. I'm thinking to maybe introduce functional options for it or maybe a struct which contain all the parameters. WDYT?

Yeah I think I'd go with something like a FranzReaderHooks type, where you can do stuff like NewFranzReaderHooks().WithRecordToMessageFn(foo).WithCloseHookFn(bar).

Signed-off-by: Mihai Todor <[email protected]>

- New `redpanda_migrator_offsets` input - Fields `offset_topic`, `offset_group`, `offset_partition`, `offset_commit_timestamp` and `offset_metadata` added to the `redpanda_migrator_offsets` output Signed-off-by: Mihai Todor <[email protected]>

Signed-off-by: Mihai Todor <[email protected]>

This is required in order to pull in twmb/franz-go#838 This is needed because the `redpanda_migrator` input needs to create all the matched topics during the first call to `ReadBatch()`. Signed-off-by: Mihai Todor <[email protected]>

Signed-off-by: Mihai Todor <[email protected]>

rockwotj · 2025-01-03T18:28:24Z

internal/impl/kafka/franz_reader_ordered.go

@@ -114,8 +121,13 @@ func (f *FranzReaderOrdered) recordsToBatch(records []*kgo.Record) *batchWithRec
 	var length uint64
 	var batch service.MessageBatch
 	for _, r := range records {
+		record, err := f.recordToMessageFn(r)
+		if err != nil {
+			f.log.Debugf("Failed to convert kafka record to message: %s", err)


This is dropping records - that seems bad right?

We're intentionally dropping tombstone records. Otherwise, kafka.FranzRecordToMessageV1() doesn't return any errors, but yeah, it's easy to misuse. Alternatively, I could set an error on the message, but I'm not aware of any other input which does that and, if the idea is to drop those messages anyway, it would be wasteful to need an extra processor for that. Not sure what's the best approach here.

Maybe we should change the signature of the record to message function so that it's clearer messages are dropped in this case, so have it return (*service.Message, bool), where nil, false is an explicit instruction to discard the record as if it didn't exist.

internal/impl/kafka/franz_reader_ordered.go

rockwotj · 2025-01-03T18:59:54Z

internal/impl/kafka/franz_reader_ordered.go

@@ -76,18 +79,22 @@ type FranzReaderOrdered struct {

 // NewFranzReaderOrderedFromConfig attempts to instantiate a new
 // FranzReaderOrdered reader from a parsed config.
-func NewFranzReaderOrderedFromConfig(conf *service.ParsedConfig, res *service.Resources, optsFn func() ([]kgo.Opt, error)) (*FranzReaderOrdered, error) {
+func NewFranzReaderOrderedFromConfig(conf *service.ParsedConfig, res *service.Resources, optsFn func() ([]kgo.Opt, error), recordToMessageFn RecordToMessageFn) (*FranzReaderOrdered, error) {


Can we expand the docs to explain what recordToMessageFn is and that it can be nil? Also is there no way to do this transform after the fact? IE a mapping on the V1 message type?

Can we expand the docs to explain what recordToMessageFn is and that it can be nil?

Good idea, sorted! I am still thinking if I should change the signature of this constructor to make it easier to use. Maybe a separate refactor...

Also is there no way to do this transform after the fact? IE a mapping on the V1 message type?

For the redpanda_migrator input, I'd lose the ability to drop tombstone messages directly here, so I'd need to use a mapping processor like I explained in the other comment. I guess that's fine even if it adds a bit of perf overhead, but may be a bit annoying if someone wants to use this redpanda_migrator input directly instead of the redpanda_migrator_bundle.
For the redpanda_migrator_offsets input it gets a bit awkward, because then we'd need a custom processor which can decode both the record.Key and record.Value using franz-go APIs. We'd also lose the ability to filter out messages which contain consumer group updates about topics we don't care about, so I'd have to move the topic filter to the processor instead. It would be confusing for people who don't know they always need to attach this processor to the redpanda_migrator_offsets input.

Here is an example of what I mean by changing this via a mapping: https://gist.github.com/rockwotj/48bddf4557a5210d6fa415a1cc736090

Basically just wrap the FranzReader in another struct that implements service.BatchReader and in it's ReadBatch function do what ever filtering/transforming you need to do. No hooks required.

rockwotj · 2025-01-03T19:01:04Z

internal/impl/kafka/enterprise/redpanda_migrator_offsets_input.go

+			}
+
+			// Consume messages from the `__consumer_offsets` topic
+			d.Topics = []string{`__consumer_offsets`}


What's the plan for Redpanda Serverless, which doesn't support this?

I have a ticket for that and for now it's still something which needs to be discussed. I don't know what we can do there, but both the redpanda_migrator_offsets input and output don't currently work with RP Serverless.

rockwotj · 2025-01-03T19:02:10Z

internal/impl/kafka/enterprise/redpanda_migrator_bundle_input.tmpl.yaml

              seed_brokers: [ "127.0.0.1:9092" ]
-              topics: [ "__consumer_offsets" ]
+              topics: [ "foobar" ]


Why is this foobar?

That's the topic filter since the redpanda_migrator_offsets input reads messages from __consumer_offsets and we need to drop updates about topics which we're not interested in.

But we don't want a filter by default? Or this should be configurable? I don't understand why this is hardcoded the way it is.

rockwotj · 2025-01-03T19:15:50Z

internal/impl/kafka/franz_reader_ordered.go

@@ -80,7 +99,7 @@ type FranzReaderOrdered struct {

 // NewFranzReaderOrderedFromConfig attempts to instantiate a new
 // FranzReaderOrdered reader from a parsed config.
-func NewFranzReaderOrderedFromConfig(conf *service.ParsedConfig, res *service.Resources, optsFn func() ([]kgo.Opt, error), recordToMessageFn RecordToMessageFn) (*FranzReaderOrdered, error) {
+func NewFranzReaderOrderedFromConfig(conf *service.ParsedConfig, res *service.Resources, optsFn func() ([]kgo.Opt, error), recordToMessageFn recordToMessageFn, preflightHookFn preflightHookFn, closeHookFn func(res *service.Resources)) (*FranzReaderOrdered, error) {


I'm not really loving all these extra hooks into this method... It would be great to wrap this method/reader instead (ala decorator pattern style) - can we do that? I think the net result would be cleaner.

Yep, I hate it too (see above), but I wasn't sure how to go about it without making this PR huge.

Which of the following would you prefer:

Have exported fields on the reader and writer and set them after calling the NewFranzReaderOrderedFromConfig() and NewFranzWriterFromConfig() constructors

Use the functional options pattern

Pass a struct which contains all these parameters to the constructors

Something else
I guess you're thinking of the first approach which I haven't considered initially and the functional options approach feels overkill. Embedding the FranzReaderOrdered and FranzWriter structs in new structs and overriding the relevant methods would be nice, but then I have to introduce some more exported helper methods... Let me know if you'd like me to explore this approach.

This is what I was thinking as a diff you can apply on top of this PR: https://gist.github.com/rockwotj/48bddf4557a5210d6fa415a1cc736090. It requires exporting the kgo.Client, but I think that's an okay thing. I haven't seen any hooks yet that fundamentally cannot be implemented in a wrapper layer (decorator pattern).

internal/impl/kafka/franz_reader_ordered.go

rockwotj · 2025-01-03T19:19:23Z

internal/impl/kafka/franz_writer.go

 }

 // NewFranzWriterFromConfig uses a parsed config to extract customisation for
 // writing data to a Kafka broker. A closure function must be provided that is
 // responsible for granting access to a connected client.
-func NewFranzWriterFromConfig(conf *service.ParsedConfig, accessClientFn func(FranzSharedClientUseFn) error, yieldClientFn func(context.Context) error) (*FranzWriter, error) {
+func NewFranzWriterFromConfig(conf *service.ParsedConfig, accessClientFn accessClientFn, yieldClientFn yieldClientFn, onConnectHookFn func(client *kgo.Client), onWriteHookFn onWriteHookFn) (*FranzWriter, error) {


Same here with these hooks, can we wrap the writer that is returned instead?

internal/impl/kafka/enterprise/redpanda_migrator_offsets_input.go

Signed-off-by: Mihai Todor <[email protected]>

rockwotj · 2025-01-07T20:14:06Z

internal/impl/kafka/enterprise/redpanda_migrator_offsets_input.go

+				matchesTopic := func(topic string) bool {
+					if len(topicPatterns) > 0 {
+						return slices.ContainsFunc(topicPatterns, func(tp *regexp.Regexp) bool {
+							return tp.MatchString(topic)
+						})
+					}
+					return slices.ContainsFunc(topics, func(t string) bool {
+						return t == topic


good! Now I think this can be lifted out of this closure so it's declared before NewFranzReaderOrderedFromConfig is called. It helps remove clutter from what this function is doing.

mihaitodor force-pushed the mihaitodor-add-redpanda-migrator-offset-metadata branch from 34421d0 to 081592f Compare November 21, 2024 02:46

mihaitodor force-pushed the mihaitodor-add-redpanda-migrator-offset-metadata branch 12 times, most recently from a86bdbd to 72237c4 Compare December 12, 2024 01:18

mihaitodor force-pushed the mihaitodor-add-redpanda-migrator-offset-metadata branch 5 times, most recently from d37239f to 784ff42 Compare December 16, 2024 11:16

mihaitodor changed the title ~~Add Redpanda Migrator offset metadata~~ Refactor Redpanda Migrator components Dec 16, 2024

mihaitodor marked this pull request as ready for review December 16, 2024 11:21

mihaitodor commented Dec 16, 2024

View reviewed changes

mihaitodor force-pushed the mihaitodor-add-redpanda-migrator-offset-metadata branch from 784ff42 to 642fd09 Compare December 16, 2024 11:43

mihaitodor force-pushed the mihaitodor-add-redpanda-migrator-offset-metadata branch 6 times, most recently from 34c5d16 to 5749553 Compare December 31, 2024 14:13

mihaitodor requested review from Jeffail and removed request for Jeffail December 31, 2024 14:50

mihaitodor requested review from rockwotj, Jeffail and ooesili December 31, 2024 14:50

mihaitodor force-pushed the mihaitodor-add-redpanda-migrator-offset-metadata branch from 5749553 to bcfeae2 Compare December 31, 2024 15:07

mihaitodor commented Dec 31, 2024

View reviewed changes

mihaitodor added 17 commits January 3, 2025 17:57

Do not reject messages for which we can't perform schema ID translation

2b00f63

Signed-off-by: Mihai Todor <[email protected]>

Update redpanda_migrator input to not convert keys to string

8e1c7ee

Signed-off-by: Mihai Todor <[email protected]>

Add Redpanda Migrator offset metadata

21ac110

- New `redpanda_migrator_offsets` input - Fields `offset_topic`, `offset_group`, `offset_partition`, `offset_commit_timestamp` and `offset_metadata` added to the `redpanda_migrator_offsets` output Signed-off-by: Mihai Todor <[email protected]>

Fix error message in redpanda_migrator_offsets input

beee211

Signed-off-by: Mihai Todor <[email protected]>

Make some more changes to the integration test

c9280bb

Signed-off-by: Mihai Todor <[email protected]>

Switch the redpanda_migrator to the new ack mechanism

b6bc0b5

Signed-off-by: Mihai Todor <[email protected]>

Check for cluster connectivity in the kafka_franz input

dba9ab3

Signed-off-by: Mihai Todor <[email protected]>

Update github.com/twmb/franz-go

268050b

This is required in order to pull in twmb/franz-go#838 This is needed because the `redpanda_migrator` input needs to create all the matched topics during the first call to `ReadBatch()`. Signed-off-by: Mihai Todor <[email protected]>

Make sure we exit the preflight hook if the context is cancelled

4c4988f

Signed-off-by: Mihai Todor <[email protected]>

Cleanup redpanda_migrator_offsets input

0f23ff6

Signed-off-by: Mihai Todor <[email protected]>

Cleanup NewFranzReaderOrderedFromConfig constructor

6833a69

Signed-off-by: Mihai Todor <[email protected]>

Refactor Migrator integration test

f29972d

Signed-off-by: Mihai Todor <[email protected]>

Fix Migrator integration test

e5d5fdc

Signed-off-by: Mihai Todor <[email protected]>

Add labels to Redpanda Migrator bundle subcomponents

9ff2b99

Signed-off-by: Mihai Todor <[email protected]>

Don't emit tombstone messages from redpanda_migrator input

708b33d

Signed-off-by: Mihai Todor <[email protected]>

Update info.csv

0683073

Signed-off-by: Mihai Todor <[email protected]>

Update changelog

e0838a5

Signed-off-by: Mihai Todor <[email protected]>

mihaitodor force-pushed the mihaitodor-add-redpanda-migrator-offset-metadata branch from bcfeae2 to e0838a5 Compare January 3, 2025 18:29

rockwotj reviewed Jan 3, 2025

View reviewed changes

Address review feedback

5ff42a4

Signed-off-by: Mihai Todor <[email protected]>

mihaitodor requested a review from rockwotj January 5, 2025 21:23

rockwotj reviewed Jan 7, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactor Redpanda Migrator components #3026

Refactor Redpanda Migrator components #3026

mihaitodor commented Nov 21, 2024 •

edited

Loading

mihaitodor Dec 16, 2024 •

edited

Loading

rockwotj Jan 3, 2025

mihaitodor Jan 4, 2025

Jeffail Jan 8, 2025

mihaitodor Dec 31, 2024

Jeffail Jan 8, 2025

rockwotj Jan 3, 2025

mihaitodor Jan 4, 2025

Jeffail Jan 8, 2025

rockwotj Jan 3, 2025

mihaitodor Jan 4, 2025 •

edited

Loading

rockwotj Jan 7, 2025

rockwotj Jan 3, 2025

mihaitodor Jan 4, 2025

rockwotj Jan 3, 2025

mihaitodor Jan 4, 2025

rockwotj Jan 7, 2025

rockwotj Jan 3, 2025

mihaitodor Jan 5, 2025

rockwotj Jan 7, 2025

rockwotj Jan 3, 2025

rockwotj Jan 7, 2025

Refactor Redpanda Migrator components #3026

Are you sure you want to change the base?

Refactor Redpanda Migrator components #3026

Conversation

mihaitodor commented Nov 21, 2024 • edited Loading

Added

Fixed

Changed

Redpanda Migrator offset metadata

mihaitodor Dec 16, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mihaitodor Jan 4, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mihaitodor commented Nov 21, 2024 •

edited

Loading

mihaitodor Dec 16, 2024 •

edited

Loading

mihaitodor Jan 4, 2025 •

edited

Loading