add log message with stacktrace on close (#113)

add log message with stacktrace on close

streamclient/kafkaclient/consumer.go

@@ -111,6 +111,8 @@ func (c *consumer) Nack(m stream.Message) error {
 // but it will only effectively close the consumer on the first call.
 func (c *consumer) Close() (err error) {
 	c.once.Do(func() {
+		c.logger.Info("Request received to close consumer.", zap.Stack("stacktrace"))
+
 		// This synchronous call closes the Kafka consumer and also sends any
 		// still-to-be-committed offsets to the Broker before returning. This is
 		// done first, so that no new messages are delivered to us, before we close

streamclient/kafkaclient/producer.go

@@ -102,6 +102,8 @@ func (p *producer) Errors() <-chan error {
 // producer on the first call.
 func (p *producer) Close() (err error) {
 	p.once.Do(func() {
+		p.logger.Info("Request received to close producer.", zap.Stack("stacktrace"))
+
 		// Trigger the quit channel, which terminates our internal goroutine to
 		// process messages, and closes the messages channel. We do this first, to
 		// prevent sending any left-over messages to a closed rdkafka producer