Revert "Fixes for consumer group (#1022)"

This reverts commit a5f270d.

Author: rhansen2

consumergroup.go

@@ -432,15 +432,8 @@ func (g *Generation) CommitOffsets(offsets map[string]map[int]int64) error {
 		Topics:       topics,
 	}
 
-	resp, err := g.coord.offsetCommit(genCtx{g}, request)
+	_, err := g.coord.offsetCommit(genCtx{g}, request)
 	if err == nil {
-		for _, partitions := range resp.Topics {
-			for _, partition := range partitions {
-				if partition.Error != nil {
-					return partition.Error
-				}
-			}
-		}
 		// if logging is enabled, print out the partitions that were committed.
 		g.log(func(l Logger) {
 			var report []string
@@ -477,17 +470,14 @@ func (g *Generation) heartbeatLoop(interval time.Duration) {
 			case <-ctx.Done():
 				return
 			case <-ticker.C:
-				resp, err := g.coord.heartbeat(ctx, &HeartbeatRequest{
+				_, err := g.coord.heartbeat(ctx, &HeartbeatRequest{
 					GroupID:      g.GroupID,
 					GenerationID: g.ID,
 					MemberID:     g.MemberID,
 				})
 				if err != nil {
 					return
 				}
-				if resp.Error != nil {
-					return
-				}
 			}
 		}
 	})
@@ -1101,9 +1091,6 @@ func (cg *ConsumerGroup) fetchOffsets(subs map[string][]int) (map[string]map[int
 	for topic, offsets := range offsets.Topics {
 		offsetsByPartition := map[int]int64{}
 		for _, pr := range offsets {
-			if pr.Error != nil {
-				return nil, pr.Error
-			}
 			if pr.CommittedOffset < 0 {
 				pr.CommittedOffset = cg.config.StartOffset
 			}
@@ -1150,17 +1137,14 @@ func (cg *ConsumerGroup) leaveGroup(ctx context.Context, memberID string) error
 		log.Printf("Leaving group %s, member %s", cg.config.ID, memberID)
 	})
 
-	resp, err := cg.coord.leaveGroup(ctx, &LeaveGroupRequest{
+	_, err := cg.coord.leaveGroup(ctx, &LeaveGroupRequest{
 		GroupID: cg.config.ID,
 		Members: []LeaveGroupRequestMember{
 			{
 				ID: memberID,
 			},
 		},
 	})
-	if err == nil && resp.Error != nil {
-		err = resp.Error
-	}
 	if err != nil {
 		cg.withErrorLogger(func(log Logger) {
 			log.Printf("leave group failed for group, %v, and member, %v: %v", cg.config.ID, memberID, err)

consumergroup_test.go

@@ -606,96 +606,3 @@ func TestGenerationStartsFunctionAfterClosed(t *testing.T) {
 		}
 	}
 }
-
-func TestGenerationEndsOnHeartbeatError(t *testing.T) {
-	gen := Generation{
-		coord: &mockCoordinator{
-			heartbeatFunc: func(context.Context, *HeartbeatRequest) (*HeartbeatResponse, error) {
-				return nil, errors.New("some error")
-			},
-		},
-		done:     make(chan struct{}),
-		joined:   make(chan struct{}),
-		log:      func(func(Logger)) {},
-		logError: func(func(Logger)) {},
-	}
-
-	ch := make(chan error)
-	gen.Start(func(ctx context.Context) {
-		<-ctx.Done()
-		ch <- ctx.Err()
-	})
-
-	gen.heartbeatLoop(time.Millisecond)
-
-	select {
-	case <-time.After(time.Second):
-		t.Fatal("timed out waiting for func to run")
-	case err := <-ch:
-		if !errors.Is(err, ErrGenerationEnded) {
-			t.Fatalf("expected %v but got %v", ErrGenerationEnded, err)
-		}
-	}
-}
-
-func TestGenerationEndsOnHeartbeatRebalaceInProgress(t *testing.T) {
-	gen := Generation{
-		coord: &mockCoordinator{
-			heartbeatFunc: func(context.Context, *HeartbeatRequest) (*HeartbeatResponse, error) {
-				return &HeartbeatResponse{
-					Error: makeError(int16(RebalanceInProgress), ""),
-				}, nil
-			},
-		},
-		done:     make(chan struct{}),
-		joined:   make(chan struct{}),
-		log:      func(func(Logger)) {},
-		logError: func(func(Logger)) {},
-	}
-
-	ch := make(chan error)
-	gen.Start(func(ctx context.Context) {
-		<-ctx.Done()
-		ch <- ctx.Err()
-	})
-
-	gen.heartbeatLoop(time.Millisecond)
-
-	select {
-	case <-time.After(time.Second):
-		t.Fatal("timed out waiting for func to run")
-	case err := <-ch:
-		if !errors.Is(err, ErrGenerationEnded) {
-			t.Fatalf("expected %v but got %v", ErrGenerationEnded, err)
-		}
-	}
-}
-
-func TestGenerationOffsetCommitErrorsAreReturned(t *testing.T) {
-	mc := mockCoordinator{
-		offsetCommitFunc: func(context.Context, *OffsetCommitRequest) (*OffsetCommitResponse, error) {
-			return &OffsetCommitResponse{
-				Topics: map[string][]OffsetCommitPartition{
-					"topic": {
-						{
-							Error: ErrGenerationEnded,
-						},
-					},
-				},
-			}, nil
-		},
-	}
-	gen := Generation{
-		coord: mc,
-		log:   func(func(Logger)) {},
-	}
-
-	err := gen.CommitOffsets(map[string]map[int]int64{
-		"topic": {
-			0: 100,
-		},
-	})
-	if err == nil {
-		t.Fatal("got nil from CommitOffsets when expecting an error")
-	}
-}

joingroup.go

@@ -3,9 +3,7 @@ package kafka
 import (
 	"bufio"
 	"context"
-	"errors"
 	"fmt"
-	"io"
 	"net"
 	"time"
 
@@ -165,9 +163,7 @@ func (c *Client) JoinGroup(ctx context.Context, req *JoinGroupRequest) (*JoinGro
 
 	for _, member := range r.Members {
 		var meta consumer.Subscription
-		metaVersion := makeInt16(member.Metadata[0:2])
-		err = protocol.Unmarshal(member.Metadata, metaVersion, &meta)
-		err = joinGroupSubscriptionMetaError(err, metaVersion)
+		err = protocol.Unmarshal(member.Metadata, consumer.MaxVersionSupported, &meta)
 		if err != nil {
 			return nil, fmt.Errorf("kafka.(*Client).JoinGroup: %w", err)
 		}
@@ -192,16 +188,6 @@ func (c *Client) JoinGroup(ctx context.Context, req *JoinGroupRequest) (*JoinGro
 	return res, nil
 }
 
-// sarama indicates there are some misbehaving clients out there that
-// set the version as 1 but don't include the OwnedPartitions section
-// https://github.com/Shopify/sarama/blob/610514edec1825240d59b62e4d7f1aba4b1fa000/consumer_group_members.go#L43
-func joinGroupSubscriptionMetaError(err error, version int16) error {
-	if version >= 1 && errors.Is(err, io.ErrUnexpectedEOF) {
-		return nil
-	}
-	return err
-}
-
 type groupMetadata struct {
 	Version  int16
 	Topics   []string

joingroup_test.go

@@ -5,14 +5,10 @@ import (
 	"bytes"
 	"context"
 	"errors"
-	"net"
 	"reflect"
 	"testing"
 	"time"
 
-	"github.com/segmentio/kafka-go/protocol"
-	"github.com/segmentio/kafka-go/protocol/consumer"
-	"github.com/segmentio/kafka-go/protocol/joingroup"
 	ktesting "github.com/segmentio/kafka-go/testing"
 )
 
@@ -128,84 +124,6 @@ func TestClientJoinGroup(t *testing.T) {
 	}
 }
 
-type roundTripFn func(context.Context, net.Addr, Request) (Response, error)
-
-func (f roundTripFn) RoundTrip(ctx context.Context, addr net.Addr, req Request) (Response, error) {
-	return f(ctx, addr, req)
-}
-
-// https://github.com/Shopify/sarama/blob/610514edec1825240d59b62e4d7f1aba4b1fa000/consumer_group_members.go#L43
-func TestClientJoinGroupSaramaCompatibility(t *testing.T) {
-	subscription := consumer.Subscription{
-		Version: 1,
-		Topics:  []string{"topic"},
-	}
-
-	// Marhsal as Verzon 0 (Without OwnedPartitions) but
-	// with Version=1.
-	metadata, err := protocol.Marshal(0, subscription)
-	if err != nil {
-		t.Fatalf("failed to marshal subscription %v", err)
-	}
-
-	client := &Client{
-		Addr: TCP("fake:9092"),
-		Transport: roundTripFn(func(context.Context, net.Addr, Request) (Response, error) {
-			resp := joingroup.Response{
-				ProtocolType: "consumer",
-				ProtocolName: RoundRobinGroupBalancer{}.ProtocolName(),
-				LeaderID:     "member",
-				MemberID:     "member",
-				Members: []joingroup.ResponseMember{
-					{
-						MemberID: "member",
-						Metadata: metadata,
-					},
-				},
-			}
-			return &resp, nil
-		}),
-	}
-
-	expResp := JoinGroupResponse{
-		ProtocolName: RoundRobinGroupBalancer{}.ProtocolName(),
-		ProtocolType: "consumer",
-		LeaderID:     "member",
-		MemberID:     "member",
-		Members: []JoinGroupResponseMember{
-			{
-				ID: "member",
-				Metadata: GroupProtocolSubscription{
-					Topics:          []string{"topic"},
-					OwnedPartitions: map[string][]int{},
-				},
-			},
-		},
-	}
-
-	gotResp, err := client.JoinGroup(context.Background(), &JoinGroupRequest{
-		GroupID:      "group",
-		MemberID:     "member",
-		ProtocolType: "consumer",
-		Protocols: []GroupProtocol{
-			{
-				Name: RoundRobinGroupBalancer{}.ProtocolName(),
-				Metadata: GroupProtocolSubscription{
-					Topics:   []string{"topic"},
-					UserData: metadata,
-				},
-			},
-		},
-	})
-	if err != nil {
-		t.Fatalf("error calling JoinGroup: %v", err)
-	}
-
-	if !reflect.DeepEqual(expResp, *gotResp) {
-		t.Fatalf("unexpected JoinGroup resp\nexpected: %#v\n     got: %#v", expResp, *gotResp)
-	}
-}
-
 func TestSaramaCompatibility(t *testing.T) {
 	var (
 		// sample data from github.com/Shopify/sarama

protocol/heartbeat/heartbeat.go

@@ -27,8 +27,8 @@ type Response struct {
 	// type.
 	_ struct{} `kafka:"min=v4,max=v4,tag"`
 
-	ThrottleTimeMs int32 `kafka:"min=v1,max=v4"`
 	ErrorCode      int16 `kafka:"min=v0,max=v4"`
+	ThrottleTimeMs int32 `kafka:"min=v1,max=v4"`
 }
 
 func (r *Response) ApiKey() protocol.ApiKey {

reader_test.go

@@ -1684,7 +1684,7 @@ func TestConsumerGroupMultipleWithDefaultTransport(t *testing.T) {
 		recvErr2 <- err
 	}()
 
-	time.Sleep(conf1.MaxWait * 5)
+	time.Sleep(conf1.MaxWait)
 
 	totalMessages := 10
 

syncgroup.go

@@ -127,13 +127,9 @@ func (c *Client) SyncGroup(ctx context.Context, req *SyncGroupRequest) (*SyncGro
 	r := m.(*syncgroup.Response)
 
 	var assignment consumer.Assignment
-	var metaVersion int16
-	if len(r.Assignments) > 2 {
-		metaVersion = makeInt16(r.Assignments[0:2])
-		err = protocol.Unmarshal(r.Assignments, metaVersion, &assignment)
-		if err != nil {
-			return nil, fmt.Errorf("kafka.(*Client).SyncGroup: %w", err)
-		}
+	err = protocol.Unmarshal(r.Assignments, consumer.MaxVersionSupported, &assignment)
+	if err != nil {
+		return nil, fmt.Errorf("kafka.(*Client).SyncGroup: %w", err)
 	}
 
 	res := &SyncGroupResponse{