Function TestConsumerGroupWithGroupTopicsMultiple

reader_test.go:1571–1642

(t *testing.T)

}

func TestConsumerGroupWithGroupTopicsMultiple(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()

	client, shutdown := newLocalClient()
	defer shutdown()
	t1 := makeTopic()
	createTopic(t, t1, 1)
	defer deleteTopic(t, t1)
	t2 := makeTopic()
	createTopic(t, t2, 1)
	defer deleteTopic(t, t2)
	conf := ReaderConfig{
		Brokers:                []string{"localhost:9092"},
		GroupID:                makeGroupID(),
		GroupTopics:            []string{t1, t2},
		MaxWait:                time.Second,
		PartitionWatchInterval: 100 * time.Millisecond,
		WatchPartitionChanges:  true,
		Logger:                 newTestKafkaLogger(t, "Reader:"),
	}

	r := NewReader(conf)

	w := &Writer{
		Addr:         TCP(r.config.Brokers...),
		BatchTimeout: 10 * time.Millisecond,
		BatchSize:    1,
		Transport:    client.Transport,
		Logger:       newTestKafkaLogger(t, "Writer:"),
	}
	defer w.Close()

	time.Sleep(time.Second)

	msgs := make([]Message, 0, len(conf.GroupTopics))
	for _, topic := range conf.GroupTopics {
		msgs = append(msgs, Message{Topic: topic})
	}
	if err := w.WriteMessages(ctx, msgs...); err != nil {
		t.Logf("write error: %+v", err)
	}

	wg := new(sync.WaitGroup)
	wg.Add(len(msgs))

	go func() {
		wg.Wait()
		t.Log("closing reader")
		r.Close()
	}()

	for {
		msg, err := r.ReadMessage(ctx)
		if err != nil {
			if errors.Is(err, io.EOF) {
				t.Log("reader closed")
				break