Function TestClientDeleteOffset

offsetdelete_test.go:14–160

(t *testing.T)

)

func TestClientDeleteOffset(t *testing.T) {
	if !ktesting.KafkaIsAtLeast("2.4.0") {
		return
	}

	topic := makeTopic()
	client, shutdown := newLocalClientWithTopic(topic, 3)
	defer shutdown()
	now := time.Now()

	const N = 10 * 3
	records := make([]Record, 0, N)
	for i := 0; i < N; i++ {
		records = append(records, Record{
			Time:  now,
			Value: NewBytes([]byte("test-message-" + strconv.Itoa(i))),
		})
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
	defer cancel()
	res, err := client.Produce(ctx, &ProduceRequest{
		Topic:        topic,
		RequiredAcks: RequireAll,
		Records:      NewRecordReader(records...),
	})
	if err != nil {
		t.Fatal(err)
	}

	if res.Error != nil {
		t.Error(res.Error)
	}

	for index, err := range res.RecordErrors {
		t.Fatalf("record at index %d produced an error: %v", index, err)
	}
	ctx, cancel = context.WithTimeout(context.Background(), time.Second*30)
	defer cancel()
	groupID := makeGroupID()

	group, err := NewConsumerGroup(ConsumerGroupConfig{
		ID:                groupID,
		Topics:            []string{topic},
		Brokers:           []string{"localhost:9092"},
		HeartbeatInterval: 2 * time.Second,
		RebalanceTimeout:  2 * time.Second,
		RetentionTime:     time.Hour,
		Logger:            log.New(os.Stdout, "cg-test: ", 0),
	})
	if err != nil {
		t.Fatal(err)
	}

	gen, err := group.Next(ctx)
	if err != nil {
		t.Fatal(err)
	}