Publish offsets before unassign

This is the only place where we can do it (esp. on shutdown), since it
calls `consumer.position()`, with no param, and thus updates the position
of assigned partitions.

This however has an unfortunate effect: it puts Zookeeper in the path,
as the unassign with be delayed until we have retrieved the offsets
and written them to zookeeper.

This should however be fine in the context of a kafka rebalance, which
takes seconds or more.

Issue: BB-455

lib/BackbeatConsumer.js

@@ -585,15 +585,24 @@ class BackbeatConsumer extends EventEmitter {
                     logger.bind(this._log)('rdkafka.commit failed', { e: e.toString(), assignment });
                 }
 
-                try {
-                    this._consumer.unassign();
-                } catch (e) {
-                    // Ignore exceptions if we are not connected
-                    const logger = this._consumer.isConnected() ? this._log.error : this._log.info;
-                    logger.bind(this._log)('rdkafka.unassign failed', { e: e.toString(), assignment });
-                }
+                const doUnassign = () => {
+                    try {
+                        this._consumer.unassign();
+                    } catch (e) {
+                        // Ignore exceptions if we are not connected
+                        const logger = this._consumer.isConnected() ? this._log.error : this._log.info;
+                        logger.bind(this._log)('rdkafka.unassign failed', { e: e.toString(), assignment });
+                    }
 
-                this.emit('unassign', status);
+                    this.emit('unassign', status);
+                };
+
+                // publish offsets to zookeeper
+                if (this._kafkaBacklogMetricsConfig) {
+                    this._publishOffsetsCron(doUnassign);
+                } else {
+                    doUnassign();
+                }
             });
 
             if (!this._processingQueue || this._processingQueue.idle()) {
@@ -911,37 +920,30 @@ class BackbeatConsumer extends EventEmitter {
         this._circuitBreaker.stop();
         return async.waterfall([
             next => {
-                if (this._kafkaBacklogMetricsConfig) {
-                    // publish offsets to zookeeper
-                    return this._publishOffsetsCron(() => next());
+                if (this._consumer.isConnected()) {
+                    const subscription = this._consumer.subscription() || [];
+                    if (subscription.length > 0) {
+                        this._consumer.unsubscribe();
+                        // Wait for partition unassign to complete before
+                        // disconnecting, the rebalance callback will handle
+                        // waiting for current jobs to complete as well as commit
+                        // the latest offsets
+                        this.on('unassign', () => next());
+                        return;
+                    }
                 }
-                return process.nextTick(next);
+                process.nextTick(next);
             },
             next => {
                 if (this._zookeeper) {
                     this._zookeeper.close();
-                } else {
-                    process.nextTick(next);
-                }
-            },
-            next => {
-                const subscription = this._consumer.subscription() || [];
-                if (subscription.length > 0) {
-                    this._consumer.unsubscribe();
-                    // Wait for partition unassign to complete before
-                    // disconnecting, the rebalance callback will handle
-                    // waiting for current jobs to complete as well as commit
-                    // the latest offsets
-                    this.on('unassign', next);
-                } else {
-                    next(UNASSIGN_STATUS.IDLE);
                 }
-            },
-            (status, next) => {
-                if (status !== UNASSIGN_STATUS.TIMEOUT) {
+                if (this._consumer.isConnected()) {
                     this._consumer.disconnect();
+                    this._consumer.once('disconnected', () => next());
+                } else {
+                    process.nextTick(next);
                 }
-                this._consumer.once('disconnected', () => next());
             },
         ], () => cb());
     }