kgo source: kill the session if FetchResponse is version < 7

If we did not kill the session, we would still use sessions even though
kafka does not support it. If kafka replied with a partition error, we
would stop requesting the partition and be unable to make progress.

pkg/kgo/source.go

@@ -575,7 +575,11 @@ func (s *source) fetch(consumerSession *consumerSession, doneFetch chan<- struct
 		return
 	}
 
-	if resp.SessionID > 0 {
+	if resp.Version < 7 {
+		// If the version is less than 7, we cannot use fetch sessions,
+		// so we kill them on the first response.
+		s.session.kill()
+	} else if resp.SessionID > 0 {
 		s.session.bumpEpoch(resp.SessionID)
 	}