File tree

Original file line number	Diff line number	Diff line change
`@@ -125,8 +125,8 @@ func (c *consumer) Consume(ctx context.Context) (<-chan async.Message, <-chan er`
`125`	`125`	`closeConsumer(c.cg)`
`126`	`126`	`return`
`127`	`127`	`case consumerError := <-c.cg.Errors():`
`128`		`-closeConsumer(c.cg)`
`129`	`128`	`chErr <- consumerError`
	`129`	`+closeConsumer(c.cg)`
`130`	`130`	`return`
`131`	`131`	`}`
`132`	`132`	`}`

Original file line number	Diff line number	Diff line change
`@@ -71,6 +71,7 @@ func TestGroupConsume_ClaimMessageError(t *testing.T) {`
`71`	`71`	`chErr := make(chan error)`
`72`	`72`	`go func() {`
`73`	`73`
	`74`	`+// Consumer will error out in ClaimMessage as no DecoderFunc has been set`
`74`	`75`	`factory, err := group.New("test1", uuid.New().String(), []string{groupTopic2}, Brokers(),`
`75`	`76`	`kafka.Version(sarama.V2_1_0_0.String()), kafka.StartFromNewest())`
`76`	`77`	`if err != nil {`
`@@ -103,8 +104,10 @@ func TestGroupConsume_ClaimMessageError(t *testing.T) {`
`103`	`104`
`104`	`105`	`select {`
`105`	`106`	`case <-chMessages:`
`106`		`-require.Fail(t, "no messages where expected")`
	`107`	`+require.Fail(t, "no messages were expected")`
`107`	`108`	`case err = <-chErr:`
`108`		`-require.EqualError(t, err, "kafka: tried to use a consumer group that was closed")`
	`109`	`+require.EqualError(t, err, "kafka: error while consuming groupTopic2/0: "+`
	`110`	`+"could not determine decoder failed to determine content type from message headers [] : "+`
	`111`	`+"content type header is missing")`
`109`	`112`	`}`
`110`	`113`	`}`