Function TestClientSyncGroup

syncgroup_test.go:14–157

(t *testing.T)

)

func TestClientSyncGroup(t *testing.T) {
	// In order to get to a sync group call we need to first
	// join a group.
	topic := makeTopic()
	client, shutdown := newLocalClient()
	client.Timeout = time.Minute
	// Although at higher api versions ClientID is nullable
	// for some reason the SyncGroup API call errors
	// when ClientID is null.
	// The Java Kafka Consumer generates a ClientID if one is not
	// present or if the provided ClientID is empty.
	client.Transport.(*Transport).ClientID = "test-client"
	defer shutdown()

	err := clientCreateTopic(client, topic, 3)
	if err != nil {
		t.Fatal(err)
	}

	groupID := makeGroupID()

	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()
	respc, err := waitForCoordinatorIndefinitely(ctx, client, &FindCoordinatorRequest{
		Addr:    client.Addr,
		Key:     groupID,
		KeyType: CoordinatorKeyTypeConsumer,
	})
	if err != nil {
		t.Fatal(err)
	}

	if respc.Error != nil {
		t.Fatal(err)
	}

	groupInstanceID := "group-instance-id"
	userData := "user-data"

	var rrGroupBalancer RoundRobinGroupBalancer

	req := &JoinGroupRequest{
		GroupID:          groupID,
		GroupInstanceID:  groupInstanceID,
		ProtocolType:     "consumer",
		SessionTimeout:   time.Minute,
		RebalanceTimeout: time.Minute,
		Protocols: []GroupProtocol{
			{
				Name: rrGroupBalancer.ProtocolName(),
				Metadata: GroupProtocolSubscription{
					Topics:   []string{topic},
					UserData: []byte(userData),
					OwnedPartitions: map[string][]int{
						topic: {0, 1, 2},
					},
				},
			},