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).

Nice, thanks! I was able to rework the reader to avoid introducing hooks, but might end up using this for the writer. Will think about it, but I'm pretty OK with how it looks now.

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-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.

Answered on Slack. TL; DR: This is the test section of the template where I'm asserting that the topic filter for redpanda_migrator_offsets is set to a specific topic. We don't want to send consumer group updates for all topics to the redpanda_migrator_offsets output.

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).

Very cool! Indeed, that does do the trick for the reader. I didn't think I can leverage this pattern initially, but, with enough patience, it does work out nicely :)

For the writer I'll have to do some refactoring to get rid of those hooks, but not sure if there's a clean way to avoid using a hook in WriteBatch unless I add a separate API method that's called in WriteBatch and then I can override that one. I'll think about it.

internal/impl/kafka/franz_reader_ordered.go

internal/impl/kafka/franz_writer.go

internal/impl/kafka/enterprise/redpanda_migrator_offsets_input.go

Signed-off-by: Mihai Todor <todormihai@gmail.com>

- 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 <todormihai@gmail.com>

Signed-off-by: Mihai Todor <todormihai@gmail.com>

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 <todormihai@gmail.com>

Signed-off-by: Mihai Todor <todormihai@gmail.com>

- Move OnConnect topic creation logic to the output to avoid the circular dependency between the input and output (the input doesn't need to know about the output) - Clean up error handling Signed-off-by: Mihai Todor <todormihai@gmail.com>

Signed-off-by: Mihai Todor <todormihai@gmail.com>

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 commented Dec 31, 2024

View reviewed changes

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

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

rockwotj reviewed Jan 7, 2025

View reviewed changes

internal/impl/kafka/enterprise/redpanda_migrator_offsets_input.go Outdated Show resolved Hide resolved

mihaitodor force-pushed the mihaitodor-add-redpanda-migrator-offset-metadata branch 2 times, most recently from 7888a52 to 9466860 Compare January 13, 2025 03:23

mihaitodor added 22 commits January 13, 2025 03:27

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

c16d957

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Update redpanda_migrator input to not convert keys to string

fbf9d6e

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Add Redpanda Migrator offset metadata

d080771

- 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 <todormihai@gmail.com>

Fix error message in redpanda_migrator_offsets input

8e5fb61

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Make some more changes to the integration test

47b157f

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Switch the redpanda_migrator to the new ack mechanism

f7224cd

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Check for cluster connectivity in the kafka_franz input

00d9d32

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Update github.com/twmb/franz-go

84d65be

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 <todormihai@gmail.com>

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

e2f8db7

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Cleanup redpanda_migrator_offsets input

5a0d852

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Cleanup NewFranzReaderOrderedFromConfig constructor

a36eedc

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Refactor Migrator integration test

cfb6646

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Fix Migrator integration test

fab411e

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Add labels to Redpanda Migrator bundle subcomponents

1bfa3a6

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Don't emit tombstone messages from redpanda_migrator input

ff69b24

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Update info.csv

ce878bb

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Address review feedback

bae41ea

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Refactor matchesTopic helper as suggested by Tyler

843b99a

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Simplify Migrator logic

1c863ab

- Move OnConnect topic creation logic to the output to avoid the circular dependency between the input and output (the input doesn't need to know about the output) - Clean up error handling Signed-off-by: Mihai Todor <todormihai@gmail.com>

Update changelog

853c73b

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Rework recordToMessageFn to return bool instead of error

125149c

Signed-off-by: Mihai Todor <todormihai@gmail.com>

Remove hooks from FranzReaderOrdered

959a5db

Signed-off-by: Mihai Todor <todormihai@gmail.com>

mihaitodor force-pushed the mihaitodor-add-redpanda-migrator-offset-metadata branch from 9466860 to 959a5db Compare January 13, 2025 03:28

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

mihaitodor Jan 13, 2025

rockwotj Jan 3, 2025

mihaitodor Jan 4, 2025

Jeffail Jan 8, 2025

rockwotj Jan 3, 2025

mihaitodor Jan 4, 2025

rockwotj Jan 3, 2025

mihaitodor Jan 4, 2025

rockwotj Jan 7, 2025

mihaitodor Jan 13, 2025

rockwotj Jan 3, 2025

mihaitodor Jan 5, 2025

rockwotj Jan 7, 2025

mihaitodor Jan 13, 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

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