Added consumer shutdown for embedded kafka consumer.