Function TestConsumerGroupWithTopic

reader_test.go:1465–1515

(t *testing.T)

}

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

	conf := ReaderConfig{
		Brokers:                []string{"localhost:9092"},
		GroupID:                makeGroupID(),
		Topic:                  makeTopic(),
		MaxWait:                time.Second,
		PartitionWatchInterval: 100 * time.Millisecond,
		WatchPartitionChanges:  true,
		Logger:                 newTestKafkaLogger(t, "Reader:"),
	}

	r := NewReader(conf)
	defer r.Close()

	recvErr := make(chan error, len(conf.GroupTopics))
	go func() {
		msg, err := r.ReadMessage(ctx)
		t.Log(msg)
		recvErr <- err
	}()

	time.Sleep(conf.MaxWait)

	client, shutdown := newLocalClientWithTopic(conf.Topic, 1)
	defer shutdown()

	w := &Writer{
		Addr:         TCP(r.config.Brokers...),
		Topic:        conf.Topic,
		BatchTimeout: 10 * time.Millisecond,
		BatchSize:    1,
		Transport:    client.Transport,
		Logger:       newTestKafkaLogger(t, "Writer:"),
	}
	defer w.Close()
	if err := w.WriteMessages(ctx, Message{Value: []byte(conf.Topic)}); err != nil {
		t.Fatalf("write error: %+v", err)
	}

	if err := <-recvErr; err != nil {
		t.Fatalf("read error: %+v", err)
	}

	nMsgs := r.Stats().Messages
	if nMsgs != 1 {
		t.Fatalf("expected to receive 1 message, but got %d", nMsgs)
	}
}

func TestConsumerGroupWithGroupTopicsSingle(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)