source: only bump session epoch if session id is non-zero

Otherwise, we inadvertently bump epochs for killed sessions, which results in the epoch quickly becoming invalid.
twmb · Jan 29, 2021 · c7caea1 · c7caea1
1 parent a448e82
commit c7caea1
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/pkg/kgo/source.go b/pkg/kgo/source.go
@@ -537,7 +537,9 @@ func (s *source) fetch(consumerSession *consumerSession) (fetched bool) {
 		return
 	}
 
-	s.session.bumpEpoch(resp.SessionID)
+	if resp.SessionID > 0 {
+		s.session.bumpEpoch(resp.SessionID)
+	}
 
 	// If we moved any partitions to preferred replicas, we reset the
 	// session. We do this after bumping the epoch just to ensure that we