kgo: inject the group lost error into polling

Multiple users have been hit by this and have not had visibility.

In fact, the original HookGroupManageError can be attributed to having
no visibility.

For #147, #321, #379.

pkg/kgo/consumer_group.go

@@ -349,6 +349,7 @@ func (g *groupConsumer) manage() {
 					h.OnGroupManageError(err)
 				}
 			})
+			g.c.addFakeReadyForDraining("", 0, err)
 		}
 
 		// If we are eager, we should have invalidated everything

pkg/kgo/errors.go

@@ -284,3 +284,16 @@ func (e *errUnknownCoordinator) Error() string {
 			" but did not reply with that broker in the broker list", e.key.name, e.key.typ, e.coordinator)
 	}
 }
+
+// ErrGroupSession is injected into a poll if an error occurred such that your
+// consumer group member was kicked from the group or was never able to join
+// the group.
+type ErrGroupLost struct {
+	err error
+}
+
+func (e *ErrGroupLost) Error() string {
+	return fmt.Sprintf("unable to join group: %v", e.err)
+}
+
+func (e *ErrGroupLost) Unwrap() error { return e.err }

pkg/kgo/record_and_fetch.go

@@ -324,7 +324,7 @@ type FetchError struct {
 // Errors returns all errors in a fetch with the topic and partition that
 // errored.
 //
-// There are four classes of errors possible:
+// There are a few classes of errors possible:
 //
 //  1. a normal kerr.Error; these are usually the non-retriable kerr.Errors,
 //     but theoretically a non-retriable error can be fixed at runtime (auth
@@ -345,6 +345,17 @@ type FetchError struct {
 //     is returned from every Poll call if the client has been closed.
 //     A corresponding helper function IsClientClosed can be used to detect
 //     this error.
+//
+//  5. an injected context error; this can be present if the context you were
+//     using for polling timed out or was canceled.
+//
+// If you are group consuming, there is another class of error:
+//
+//  6. an injected ErrGroupSession; this is an informational error that is
+//     injected once a group session is lost in a way that is not the standard
+//     rebalance. This error can signify that your consumer member is not able
+//     to connect to the group (ACL problems, unreachable broker), or you
+//     blocked rebalancing for too long, or your callbacks took too long.
 func (fs Fetches) Errors() []FetchError {
 	var errs []FetchError
 	fs.EachError(func(t string, p int32, err error) {