kafka-go

kafka-go copied to clipboard

Describe the bug

When using ReadMessage with IsolationLevel set to ReadCommitted, I appear to be getting back records from uncommitted transactions.

Kafka Version

Kafka 3.6/3.7

kafka-go 0.4.47

• What version(s) of Kafka are you testing against?
• What version of kafka-go are you using?

To Reproduce

I don't have exact repro steps, as I'm using Flink to produce the data - but I think this is likely reproducible be producing data in a transaction and then skipping committing it.

// snippets
	r := segment_kafka.NewReader(
		segment_kafka.ReaderConfig{
			Brokers: []string{
				"my_broker:9092",
			},
			Topic:          "my_topic",
			GroupID:        fmt.Sprintf("testing_%s", strconv.Itoa(rand.Int())),
			StartOffset:    segment_kafka.FirstOffset,
			MaxBytes:       10e6,
			IsolationLevel: segment_kafka.ReadCommitted,
		},
	)

	ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
	defer cancel()
	messages := []segment_kafka.Message{}
	for {
		select {
		case <-ctx.Done():
			return
		default:
			ev, err := r.ReadMessage(ctx)
			if err != nil {
				logger.Error("failed to read message", zap.Error(err))
				continue
			}
			logger.Info("Consumed event from topic",
				zap.String("topic", ev.Topic),
				zap.String("key", string(ev.Key)),
				zap.String("value", string(ev.Value)),
				zap.Int64("offset", ev.Offset),
				zap.Any("partition", ev.Partition),
			)
			messages = append(messages, ev)
		}
	}

Expected Behavior

I expect to only see data from committed transactions, or at a minimum have this behavior be configurable.