Add a Kafka reader

A necessary building block for OCTRL-1049

Author: knopers8

common/event/reader.go

@@ -0,0 +1,132 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2025 CERN and copyright holders of ALICE O².
+ * Author: Piotr Konopka <pkonopka@cern.ch>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * In applying this license CERN does not waive the privileges and
+ * immunities granted to it by virtue of its status as an
+ * Intergovernmental Organization or submit itself to any jurisdiction.
+ */
+
+package event
+
+import (
+	"context"
+	"fmt"
+	"github.com/AliceO2Group/Control/common/event/topic"
+	"github.com/AliceO2Group/Control/common/logger/infologger"
+	pb "github.com/AliceO2Group/Control/common/protos"
+	"github.com/segmentio/kafka-go"
+	"github.com/spf13/viper"
+	"google.golang.org/protobuf/proto"
+	"sync"
+)
+
+// Reader interface provides methods to read events.
+type Reader interface {
+	Next(ctx context.Context) (*pb.Event, error)
+	Close() error
+}
+
+// DummyReader is an implementation of Reader that returns no events.
+type DummyReader struct{}
+
+// Next returns the next event or nil if there are no more events.
+func (*DummyReader) Next(context.Context) (*pb.Event, error) { return nil, nil }
+
+// Close closes the DummyReader.
+func (*DummyReader) Close() error { return nil }
+
+// KafkaReader reads events from Kafka and provides a blocking, cancellable API to fetch events.
+// Consumption mode is chosen at creation time:
+// - latestOnly=false: consume everything (from stored offsets or beginning depending on group state)
+// - latestOnly=true: seek to latest offsets on start and only receive messages produced after start
+type KafkaReader struct {
+	*kafka.Reader
+	mu    sync.Mutex
+	topic string
+}
+
+// NewReaderWithTopic creates a KafkaReader for the provided topic and starts it.
+// If latestOnly is true the reader attempts to seek to the latest offsets on start so that
+// only new messages (produced after creation) are consumed.
+func NewReaderWithTopic(topic topic.Topic, groupID string, latestOnly bool) *KafkaReader {
+	cfg := kafka.ReaderConfig{
+		Brokers:  viper.GetStringSlice("kafkaEndpoints"),
+		Topic:    string(topic),
+		GroupID:  groupID,
+		MinBytes: 1,
+		MaxBytes: 10e7,
+	}
+
+	rk := &KafkaReader{
+		Reader: kafka.NewReader(cfg),
+		topic:  string(topic),
+	}
+
+	if latestOnly {
+		// best-effort: set offset to last so we don't replay older messages
+		if err := rk.SetOffset(kafka.LastOffset); err != nil {
+			log.WithField(infologger.Level, infologger.IL_Devel).
+				Warnf("failed to set offset to last offset: %v", err)
+		}
+	}
+
+	return rk
+}
+
+// Next blocks until the next event is available or ctx is cancelled. It returns an error when the reader is closed
+// (io.EOF) or the context is cancelled. The caller is responsible for providing a cancellable ctx.
+func (r *KafkaReader) Next(ctx context.Context) (*pb.Event, error) {
+	if r == nil {
+		return nil, fmt.Errorf("nil reader")
+	}
+
+	msg, err := r.ReadMessage(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	event, err := kafkaMessageToEvent(msg)
+	if err != nil {
+		return nil, err
+	}
+
+	return event, nil
+}
+
+// Close stops the reader.
+func (r *KafkaReader) Close() error {
+	if r == nil {
+		return nil
+	}
+	// Close the underlying kafka reader which will cause ReadMessage to return an error
+	err := r.Reader.Close()
+	if err != nil {
+		log.WithField(infologger.Level, infologger.IL_Devel).
+			Errorf("failed to close kafka reader: %v", err)
+	}
+	return err
+}
+
+func kafkaMessageToEvent(m kafka.Message) (*pb.Event, error) {
+	var evt pb.Event
+	if err := proto.Unmarshal(m.Value, &evt); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal kafka message: %w", err)
+	}
+	return &evt, nil
+}

common/event/reader_test.go

@@ -0,0 +1,48 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2025 CERN and copyright holders of ALICE O².
+ * Author: Piotr Konopka <pkonopka@cern.ch>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * In applying this license CERN does not waive the privileges and
+ * immunities granted to it by virtue of its status as an
+ * Intergovernmental Organization or submit itself to any jurisdiction.
+ */
+
+package event
+
+import (
+	pb "github.com/AliceO2Group/Control/common/protos"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/segmentio/kafka-go"
+	"google.golang.org/protobuf/proto"
+)
+
+var _ = Describe("Reader", func() {
+	When("converting kafka message to event", func() {
+		It("unmarshals protobuf payload correctly", func() {
+			e := &pb.Event{Payload: &pb.Event_CoreStartEvent{CoreStartEvent: &pb.Ev_MetaEvent_CoreStart{FrameworkId: "z"}}}
+			b, err := proto.Marshal(e)
+			Expect(err).To(BeNil())
+
+			m := kafka.Message{Value: b}
+			evt, err := kafkaMessageToEvent(m)
+			Expect(err).To(BeNil())
+			Expect(evt.GetCoreStartEvent().FrameworkId).To(Equal("z"))
+		})
+	})
+})