sticky: fix and drastically simplify isComplex detection

Before this commit, if a sticky balance with tons of disparate members
only passed in **one** topic, then the code was unable to detect if the
balance was complex, because we only detected complexity after the first
topic.

Now, we simply check if all members want to consume each topic equally.
topics2memberNums is built from topics each member is interested in, so
if any topic one member is interested in is not interested in by all
members, then we know we are complex.

Author: twmb

pkg/kgo/internal/sticky/sticky.go

@@ -404,29 +404,23 @@ func (b *balancer) assignUnassignedAndInitGraph() {
 	}
 
 	partitionPotentials := make([][]uint16, cap(b.partNames)) // for each partition, who can consume it?
-	var firstTopicMembers []uint16
 	for topic, topicMembers := range topics2memberNums {
 		for partition := int32(0); partition < b.topics[topic]; partition++ {
 			tp := topicPartition{topic, partition}
 			partNum := b.partNum(tp)
 			partitionPotentials[partNum] = topicMembers
 		}
 
-		// While building partition potentials, we can check whether
-		// all topics are consumed the same.
-		if firstTopicMembers == nil {
-			firstTopicMembers = topicMembers
-			continue
-		}
-		if b.isComplex || len(topicMembers) != len(firstTopicMembers) {
+		// If the number of members interested in this topic is not the
+		// same as the number of members in this group, then **other**
+		// members are interested in other topics and not this one, and
+		// we must go to complex balancing.
+		//
+		// We could accidentally fall into isComplex if any member is
+		// not interested in anything, but realistically we do not
+		// expect members to join with no interests.
+		if len(topicMembers) != len(b.members) {
 			b.isComplex = true
-			continue
-		}
-		for i, memberNum := range topicMembers {
-			if memberNum != firstTopicMembers[i] {
-				b.isComplex = true
-				break
-			}
 		}
 	}
 

pkg/kgo/internal/sticky/sticky_test.go

@@ -881,6 +881,22 @@ func TestImbalanced(t *testing.T) {
 			},
 		},
 
+		{
+			name: "unequal",
+			members: []GroupMember{
+				{ID: "0", Topics: []string{"0"}},
+				{ID: "1", Topics: []string{"1"}},
+			},
+			topics: map[string]int32{
+				"1": 2,
+			},
+			nsticky: 0,
+			balance: map[int]resultOptions{
+				0: {[]string{"0"}, 1},
+				2: {[]string{"1"}, 1},
+			},
+		},
+
 		//
 	} {
 		t.Run(test.name, func(t *testing.T) {