kafka: deadlock when closing consumer without canceling context

[kruskall]

kafka consumer Run method will block forever unless the context is canceled.

Trying to close the consumer with consumer.Close will block forever.

This only happens with the Kafka consumer. Pubsublite is not affected.

To reproduce: cd systemtest && go test -v -run=TestShutdown/Kafka/close

consumer will block at:

apm-queue/kafka/consumer.go

Line 189 in e48ac80

fetches := c.client.PollRecords(ctx, c.cfg.MaxPollRecords)

The close method will block trying to acquire the lock.