refactor(kafka-reader): Separates kafka constructs for better separation of concerns & reuse. #14982

owen-d · 2024-11-16T00:37:51Z

Refactor Kafka Reader and Partition Committer

This PR refactors the Kafka ingestion pipeline components to achieve better separation of concerns and cleaner interfaces. The changes split the existing Reader into three distinct components:

A ReaderIfc that handles pure Kafka interactions
A partitionCommitter that manages async offset commits (already existed, but I've refactored it to use the interface)
A ReaderService that coordinates the overall lifecycle

Key Changes

Note: As of now, this just adds 3 new files with refactored versions -- if we're happy I'll replace the existing ones, but the current structure mirrors how I developed against the prior versions.

~~Disclaimer: I haven't hooked this up to testware yet; if we're happy with this direction, I'll do that next.~~ edit: done.

New Reader Interface

Created a focused interface for Kafka operations:

type ReaderIfc interface {
    Topic() string
    Partition() int32
    ConsumerGroup() string
    FetchLastCommittedOffset(ctx context.Context) (int64, error)
    FetchPartitionOffset(ctx context.Context, position SpecialOffset) (int64, error)
    Poll(ctx context.Context) ([]Record, error)
    Commit(ctx context.Context, offset int64) error
    // Set the target offset for consumption. reads will begin from here.
    SetOffsetForConsumption(offset int64)
}

Improved Committer Design

Moved the committer to depend on ReaderIfc instead of containing Kafka logic
Maintains async commit functionality but delegates actual commits to the reader
Cleaner separation between offset management and Kafka operations
Added explicit SpecialOffset type to make partition offset fetching more type-safe

Service Lifecycle

ReaderService now coordinates between the reader and committer
Clearer separation of metrics between components
More explicit error handling throughout

Benefits

Maintainability: Responsibilities are clearly separated between Kafka operations, offset management, and service lifecycle
Flexibility: The ReaderIfc can be used independently of the service wrapper, which is how I started down this path. I'd like to reuse the underlying reader in the block-builder code.

grobinson-grafana · 2024-11-21T10:18:28Z

pkg/kafka/partition/committer_refactor.go

+				continue
+			}
+
+			if err := c.Commit(ctx, currOffset); err == nil {


Can we log the error when it's non-nil?

benclive

Nice, thanks for doing this! It is definitely an improvement. I've left a few comments to address but I'd be happy to approve once they're looked at

benclive · 2024-11-21T10:26:46Z

pkg/kafka/partition/reader_service_refactor.go

+		lastCommittedOffset = int64(KafkaStartOffset)
+	}
+
+	if lastCommittedOffset > 0 {


Just spotted this now, but this should be >= as 0 is a valid offset

benclive · 2024-11-21T10:32:11Z

pkg/kafka/partition/reader_service_refactor.go

+		}
+		lastProducedOffset = lastProducedOffset - 1 // Kafka returns the next empty offset so we must subtract 1 to get the oldest written offset.
+
+		level.Debug(logger).Log("msg", "fetched latest offset information", "partition_start_offset", partitionStartOffset, "last_produced_offset", lastProducedOffset)


Could you add lastCommittedOffset to this log line? We can then understand all the decisions made beyond this point as we frequently need to know what happened here.

benclive · 2024-11-21T10:33:44Z

pkg/kafka/partition/reader_service_refactor.go

+		// Ensure there are some records to consume. For example, if the partition has been inactive for a long
+		// time and all its records have been deleted, the partition start offset may be > 0 but there are no
+		// records to actually consume.
+		if partitionStartOffset > lastProducedOffset {


I wonder if this should also be >=. What happens if a partition is new and has had a single record written to it?

Changing this causes TestPartitionReader_ProcessCommits to fail. I'd like to leave this debugging for a future PR and let this PR simply refactor the existing structure, but not the logic.

benclive · 2024-11-21T10:38:08Z

pkg/kafka/partition/committer_refactor.go

+		return
+	}
+	// Commit has internal timeouts, so this call shouldn't block for too long
+	_ = c.Commit(context.Background(), offset)


It might be useful to add an Info log here saying what the final committed offset was.
I know we have a debug line in Commit() but it is useful to know the final commit at shutdown, even if debug logging is not enabled.

benclive · 2024-11-21T10:38:39Z

pkg/kafka/partition/committer_refactor.go

+	"go.uber.org/atomic"
+)
+
+type refactoredPartitionCommitter struct {


I'm not a big fan of "refactored..." as a name!

I think it will be removed when the old code is deleted? @owen-d

ashwanthgoli · 2024-11-21T11:18:02Z

pkg/kafka/partition/reader_refactor.go

+		level.Debug(r.logger).Log(
+			"msg", "malformed response, setting to start offset",
+		)
+		return int64(KafkaStartOffset), nil


like the other failures cases in this method, can we also return error here and let the caller decide on the offset?

I agree with you, but my goal here is to refactor the existing implementation, not change it's logic. This is a good followup PR though.

ashwanthgoli · 2024-11-21T11:26:53Z

pkg/kafka/partition/reader_service_refactor.go

+			Name: "loki_ingest_storage_reader_phase",
+			Help: "The current phase of the consumer.",
+		}, []string{"phase"}),
+		receiveDelay: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{


this is not used in the service, should we remove it?

benclive

I don't see any issues with the refactor, and I'm happy to delay any logic / logging modifications to a future PR, so approving to unblock!

scaffolding refactored kafka reads

fe5c574

owen-d requested a review from a team as a code owner November 16, 2024 00:37

pull-request-size bot added the size/XL label Nov 16, 2024

owen-d added 2 commits November 18, 2024 07:54

lint

71f1537

partition tests use refactored variants + removed duplicate metric

237eaa6

owen-d force-pushed the kafka-refactor-reader branch from e279e79 to 236fd84 Compare November 18, 2024 18:54

cleanup

5813b62

owen-d force-pushed the kafka-refactor-reader branch from 236fd84 to 5813b62 Compare November 18, 2024 18:55

grobinson-grafana approved these changes Nov 21, 2024

View reviewed changes

benclive reviewed Nov 21, 2024

View reviewed changes

ashwanthgoli reviewed Nov 21, 2024

View reviewed changes

pr feedback

bf8dbb8

pull-request-size bot added size/XXL and removed size/XL labels Nov 21, 2024

owen-d force-pushed the kafka-refactor-reader branch from 7e0604d to bf8dbb8 Compare November 21, 2024 16:21

pull-request-size bot added size/XL and removed size/XXL labels Nov 21, 2024

benclive approved these changes Nov 21, 2024

View reviewed changes

owen-d added 2 commits November 22, 2024 09:34

refactors reader svc entrypoint to align w/ existing reader usage

eae90ea

replaces old kafka partition impls w/ refactored variants

9a8af08

pull-request-size bot added size/XXL and removed size/XL labels Nov 22, 2024

lint & go mod

0eb6eab

owen-d merged commit 62e7d61 into grafana:main Nov 22, 2024
60 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(kafka-reader): Separates kafka constructs for better separation of concerns & reuse. #14982

refactor(kafka-reader): Separates kafka constructs for better separation of concerns & reuse. #14982

owen-d commented Nov 16, 2024 •

edited

Loading

grobinson-grafana Nov 21, 2024

benclive left a comment

benclive Nov 21, 2024

benclive Nov 21, 2024

benclive Nov 21, 2024

owen-d Nov 21, 2024

benclive Nov 21, 2024

benclive Nov 21, 2024

grobinson-grafana Nov 21, 2024

ashwanthgoli Nov 21, 2024

owen-d Nov 21, 2024

ashwanthgoli Nov 21, 2024

benclive left a comment

refactor(kafka-reader): Separates kafka constructs for better separation of concerns & reuse. #14982

refactor(kafka-reader): Separates kafka constructs for better separation of concerns & reuse. #14982

Conversation

owen-d commented Nov 16, 2024 • edited Loading

Refactor Kafka Reader and Partition Committer

Key Changes

New Reader Interface

Improved Committer Design

Service Lifecycle

Benefits

Choose a reason for hiding this comment

benclive left a 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

benclive left a comment

Choose a reason for hiding this comment

owen-d commented Nov 16, 2024 •

edited

Loading