Function TestFuncAdminListConsumerGroupOffsets

functional_admin_test.go:369–439

(t *testing.T)

}

func TestFuncAdminListConsumerGroupOffsets(t *testing.T) {
	checkKafkaVersion(t, "0.8.2.0")
	setupFunctionalTest(t)
	defer teardownFunctionalTest(t)

	config := NewFunctionalTestConfig()
	config.ClientID = t.Name()
	client, err := NewClient(FunctionalTestEnv.KafkaBrokerAddrs, config)
	defer safeClose(t, client)
	if err != nil {
		t.Fatal(err)
	}

	group := testFuncConsumerGroupID(t)
	consumerGroup, err := NewConsumerGroupFromClient(group, client)
	if err != nil {
		t.Fatal(err)
	}
	defer safeClose(t, consumerGroup)

	offsetMgr, _ := NewOffsetManagerFromClient(group, client)
	defer safeClose(t, offsetMgr)
	markOffset(t, offsetMgr, "test.4", 0, 2)
	offsetMgr.Commit()

	coordinator, err := client.Coordinator(group)
	if err != nil {
		t.Fatal(err)
	}

	t.Logf("coordinator broker %d", coordinator.id)

	adminClient, err := NewClusterAdminFromClient(client)
	if err != nil {
		t.Fatal(err)
	}
	{
		resp, err := adminClient.ListConsumerGroupOffsets(group, map[string][]int32{"test.4": {0, 1, 2, 3}})
		if err != nil {
			t.Fatal(err)
		}
		t.Log(spew.Sdump(resp))
	}

	brokerID := coordinator.id
	t.Cleanup(
		func() {
			if err := startDockerTestBroker(context.Background(), brokerID); err != nil {
				t.Fatal(err)
			}
		},
	)
	if err := stopDockerTestBroker(context.Background(), brokerID); err != nil {
		t.Fatal(err)
	}

	{
		resp, err := adminClient.ListConsumerGroupOffsets(group, map[string][]int32{"test.4": {0, 1, 2, 3}})