CI test fixes

Makefile

@@ -4,7 +4,7 @@ default: fmt vet errcheck test
 test:
 	echo "" > coverage.txt
 	for d in `go list ./... | grep -v vendor`; do \
-		go test -p 1 -v -timeout 300s -race -coverprofile=profile.out -covermode=atomic $$d || exit 1; \
+		go test -p 1 -v -timeout 240s -race -coverprofile=profile.out -covermode=atomic $$d || exit 1; \
 		if [ -f profile.out ]; then \
 			cat profile.out >> coverage.txt; \
 			rm profile.out; \

consumer_group.go

@@ -145,6 +145,7 @@ func (c *consumerGroup) Consume(ctx context.Context, topics []string, handler Co
 	// Ensure group is not closed
 	select {
 	case <-c.closed:
+		return ErrClosedConsumerGroup
 	default:
 	}
 
@@ -373,6 +374,12 @@ func (c *consumerGroup) leave() error {
 }
 
 func (c *consumerGroup) handleError(err error, topic string, partition int32) {
+	select {
+	case <-c.closed:
+		return
+	default:
+	}
+
 	if _, ok := err.(*ConsumerError); !ok && topic != "" && partition > -1 {
 		err = &ConsumerError{
 			Topic:     topic,
@@ -383,8 +390,8 @@ func (c *consumerGroup) handleError(err error, topic string, partition int32) {
 
 	if c.config.Consumer.Return.Errors {
 		select {
-		case <-c.closed:
 		case c.errors <- err:
+		default:
 		}
 	} else {
 		Logger.Println(err)
@@ -549,6 +556,8 @@ func (s *consumerGroupSession) consume(topic string, partition int32) {
 	select {
 	case <-s.ctx.Done():
 		return
+	case <-s.parent.closed:
+		return
 	default:
 	}
 
@@ -574,7 +583,10 @@ func (s *consumerGroupSession) consume(topic string, partition int32) {
 
 	// trigger close when session is done
 	go func() {
-		<-s.ctx.Done()
+		select {
+		case <-s.ctx.Done():
+		case <-s.parent.closed:
+		}
 		claim.AsyncClose()
 	}()
 

consumer_group_test.go

@@ -24,7 +24,7 @@ func ExampleConsumerGroup() {
 	config.Consumer.Return.Errors = true
 
 	// Start with a client
-	client, err := NewClient([]string{"localhost:9092"}, nil)
+	client, err := NewClient([]string{"localhost:9092"}, config)
 	if err != nil {
 		panic(err)
 	}

functional_consumer_group_test.go

@@ -44,7 +44,7 @@ func TestFuncConsumerGroupPartitioning(t *testing.T) {
 	m2.WaitForHandlers(5)
 
 	// shutdown M2
-	m1.AssertCleanShutdown()
+	m2.AssertCleanShutdown()
 }
 
 func TestFuncConsumerGroupExcessConsumers(t *testing.T) {