sarama: consumer group deadlock on close: consumer group shutdown hangs

@PapayaJuice yes, but the gods of concurrent programming will not praise me. The main idea is to ignore happening data race in a bad case.

done := channel
go func() {
    for not_shutdown {
       cs.Consume()
    }
    cleanup()
    close(done)
}()

CloseConsumerGroup(cs, 10 * time.Second, done)
kafkaConsumeGroupClient.Close()

func CloseConsumerGroup(cs sarama.ConsumerGroup, timeout time.Duration, wait ...<-chan struct{}) {
	done := make(chan struct{})

	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()

	// if consumer group will try to close shutdown first, it may
	// hang on waiting for a consumer group session to complete
	// when there were no partitions assigned to this node. in that
	// case s.cs.Close() will hang until cluster re-balance will
	// happen or disconnect. if that is a case, the cs Close()
	// invocation will be a data race on closed client, but at least
	// it will finish.
	go func() {
		if err := cs.Close(); err != nil {
			log.Logger().WithError(err).Info("close consumer group")
		}

		close(done)
	}()

	select {
	case <-ctx.Done():
		// consumer did not exit
		log.Logger().WithError(context.DeadlineExceeded).Info("close consumer group")
	case <-done:
		// cs.Close() finished successfully,
		// consume exited and closed `done`
		for _, ch := range wait {
			select {
			case <-ctx.Done():
				// consumer did not exit
				log.Logger().WithError(context.DeadlineExceeded).Info("close consumer group: wait external chan")
				return
			case <-ch:
				// finished
			}
		}
	}
}

The main idea behind that code is that it spawns a goroutine which tries to close a consumer group. If it deadlocks, the client which is owned by the CS will be closed forcibly in a goroutine which called CloseConsumerGroup. In a good case, the close of the CS client has a happens-before relationship set by the waiting in CloseConsumerGroup. In the bad case, it will be closed with a data race. But whatever, the cs.Close() would not block an execution thread and shutdown will continue.

The wait channels in arguments are required for this function to wait for finishing of the goroutine which does the consumer group session loop. When cs.Consume() finishes this goroutine may have a desire to do some cleanup.