sarama asyncproducer plugin - possibility of data transfer over a closed channel

[dwkang]

	go func() {
		for {
			select {
			case msgCtx, ok := <-wrapped.inputContext:
				if !ok {
					return
				}
				span := newAsyncProducerTracer(msgCtx.ctx, addrs, msgCtx.msg, config)
				producer.Input() <- msgCtx.msg
				saveAsyncProducerTracer(config, wrapped, span)

			case msg, ok := <-wrapped.input:
				if !ok {
					return
				}
				span := newAsyncProducerTracer(wrapped.ctx, addrs, msg, config)
				producer.Input() <- msg
				saveAsyncProducerTracer(config, wrapped, span)
			}
		}
	}()

If it close the async producer before the topic message is processed,
sending may occur to the input channel closed on the async producer.