Merge pull request #1263 from twmb/tfix

more ci test improvements; skipped_features update

Author: twmb

pkg/kfake/groups.go

@@ -276,7 +276,7 @@ func (gs *groups) newGroup(name string) *group {
 		pending:       make(map[string]*groupMember),
 		staticMembers: make(map[string]string),
 		protocols:     make(map[string]int),
-		reqCh:         make(chan *clientReq),
+		reqCh:         make(chan *clientReq, 16),
 		controlCh:     make(chan func(), 1), // buffer 1: holds a pending notifyTopicChange
 		quitCh:        make(chan struct{}),
 	}

pkg/kfake/skipped_features

@@ -72,13 +72,10 @@
 
 ## 68_consumer_group_heartbeat (KIP-848)
 - SubscribedTopicRegex: RE2 syntax, not Java RE2J
-- Member epoch advancement only when fully converged: kfake's conservative
-  approach is safe and simpler. Advancing in UNRELEASED_PARTITIONS state
-  only affects convergence speed, not correctness.
-- No ASSIGNING/UNRELEASED_PARTITIONS states: kfake uses empty/reconciling/stable.
-  Missing intermediate states don't affect kgo client behavior.
-- Metadata hash / metadata expiration: deliberate divergence; kfake recomputes
-  assignments on every relevant heartbeat rather than caching a metadata hash
+- Metadata hash: kfake uses notifyTopicChange() from the cluster run loop
+  instead of hashing metadata. Functionally equivalent for in-process use.
+- Group downgrade (consumer -> classic): not supported. Once a group
+  upgrades to "consumer" type, it stays that way.
 
 ## General
 - Tiered storage: Not implemented (KIP-405, KIP-1005)

pkg/kgo/consumer_test.go

@@ -860,6 +860,9 @@ func TestGroupSimple(t *testing.T) {
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
+			if tc.enable848 && !allow848 {
+				t.Skip("broker does not support KIP-848 (requires ConsumerGroupHeartbeat, key 68)")
+			}
 
 			t1, cleanup := tmpTopicPartitions(t, 1)
 			defer cleanup()

pkg/kgo/helpers_test.go

@@ -44,6 +44,9 @@ var (
 	// Static membership (KIP-345) requires JoinGroup v5+.
 	allowStaticMembership = false
 
+	// KIP-848 requires ConsumerGroupHeartbeat (key 68).
+	allow848 = false
+
 	// KGO_TEST_TLS: DSL syntax is ({ca|cert|key}:path),{1,3}
 	testCert *tls.Config
 
@@ -211,6 +214,9 @@ func init() {
 			if v, ok := versions.LookupMaxKeyVersion(11); ok && v >= 5 { // 11 = JoinGroup
 				allowStaticMembership = true
 			}
+			if _, ok := versions.LookupMaxKeyVersion(68); ok { // 68 = ConsumerGroupHeartbeat
+				allow848 = true
+			}
 			break
 		}
 		if time.Now().After(deadline) {
@@ -564,6 +570,9 @@ func testChainETL(
 	if instanceID != "" && !allowStaticMembership {
 		t.Skip("broker does not support static membership (requires JoinGroup v5+, KIP-345)")
 	}
+	if enable848 && !allow848 {
+		t.Skip("broker does not support KIP-848 (requires ConsumerGroupHeartbeat, key 68)")
+	}
 	errs := make(chan error)
 	var (
 		/////////////