[test]: make Kafka consumer read from the earliest messages (#4981)

## Which problem is this PR solving?
- Resolves #4979

## Description of the changes
- Change kafka consumer to consume messages from the oldest possible
offsets to prevent failure when the test produces messages before the
consumer is ready.

## How was this change tested?
- Start a kafka server hosted at localhost:9092
```
docker run --name kafka \
    --network jaeger \
    -p 9092:9092 \
    -e KAFKA_CFG_NODE_ID=0 \
    -e KAFKA_CFG_PROCESS_ROLES=controller,broker \
    -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093 \
    -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
    -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
    -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
    -e KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER \
    -e KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT \
    bitnami/kafka:3.6
```
- Run `./scripts/kafka-integration-test.sh`

## Checklist
- [x] I have read
https://github.com/jaegertracing/jaeger/blob/master/CONTRIBUTING_GUIDELINES.md
- [x] I have signed all commits
- [x] I have added unit tests for the new functionality
- [x] I have run lint and test steps successfully
  - for `jaeger`: `make lint test`
  - for `jaeger-ui`: `yarn lint` and `yarn test`

---------

Signed-off-by: James Ryans <james.ryans2012@gmail.com>

.github/workflows/ci-kafka.yml

@@ -62,7 +62,7 @@ jobs:
         env:
           ALLOW_ANONYMOUS_LOGIN: yes
       kafka:
-        image: bitnami/kafka:3.1.0
+        image: bitnami/kafka:3.6.0
         ports:
           - 9092:9092
         options: >-

cmd/ingester/app/builder/builder.go

@@ -53,6 +53,7 @@ func CreateConsumer(logger *zap.Logger, metricsFactory metrics.Factory, spanWrit
 	consumerConfig := kafkaConsumer.Configuration{
 		Brokers:              options.Brokers,
 		Topic:                options.Topic,
+		InitialOffset:        options.InitialOffset,
 		GroupID:              options.GroupID,
 		ClientID:             options.ClientID,
 		ProtocolVersion:      options.ProtocolVersion,

pkg/kafka/consumer/config.go

@@ -44,10 +44,11 @@ type Configuration struct {
 
 	Brokers         []string `mapstructure:"brokers"`
 	Topic           string   `mapstructure:"topic"`
-	GroupID         string   `mapstructure:"group_id"`
-	ClientID        string   `mapstructure:"client_id"`
-	ProtocolVersion string   `mapstructure:"protocol_version"`
-	RackID          string   `mapstructure:"rack_id"`
+	InitialOffset   int64
+	GroupID         string `mapstructure:"group_id"`
+	ClientID        string `mapstructure:"client_id"`
+	ProtocolVersion string `mapstructure:"protocol_version"`
+	RackID          string `mapstructure:"rack_id"`
 }
 
 // NewConsumer creates a new kafka consumer
@@ -71,5 +72,8 @@ func (c *Configuration) NewConsumer(logger *zap.Logger) (Consumer, error) {
 	// that does not set saramaConfig.Consumer.Offsets.CommitInterval to its default value 1s.
 	// then the samara-cluster fails if the default interval is not 1s.
 	saramaConfig.Consumer.Offsets.CommitInterval = time.Second
+	if c.InitialOffset != 0 {
+		saramaConfig.Consumer.Offsets.Initial = c.InitialOffset
+	}
 	return cluster.NewConsumer(c.Brokers, c.GroupID, []string{c.Topic}, saramaConfig)
 }

plugin/storage/integration/kafka_test.go

@@ -21,13 +21,15 @@ import (
 	"testing"
 	"time"
 
+	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 
 	"github.com/jaegertracing/jaeger/cmd/ingester/app"
 	"github.com/jaegertracing/jaeger/cmd/ingester/app/builder"
 	"github.com/jaegertracing/jaeger/model"
 	"github.com/jaegertracing/jaeger/pkg/config"
+	"github.com/jaegertracing/jaeger/pkg/kafka/consumer"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
 	"github.com/jaegertracing/jaeger/pkg/testutils"
 	"github.com/jaegertracing/jaeger/plugin/storage/kafka"
@@ -90,7 +92,11 @@ func (s *KafkaIntegrationTestSuite) initialize() error {
 	if err != nil {
 		return err
 	}
-	options := app.Options{}
+	options := app.Options{
+		Configuration: consumer.Configuration{
+			InitialOffset: sarama.OffsetOldest,
+		},
+	}
 	options.InitFromViper(v)
 	traceStore := memory.NewStore()
 	spanConsumer, err := builder.CreateConsumer(s.logger, metrics.NullFactory, traceStore, options)