ack() is not happening with BasicBolt / Apache Storm 1.0.3? #88

import pprint
import traceback

import config as cfg
import inflection
from db.bigtable import records
from db.bigtable.connections import connect
from petrel.emitter import BasicBolt

# enable loggers
log = cfg.config.loggers.storm

# establish bigtable connection
connect(*cfg.config.db_connection_params, admin=True)

class PersistenceBolt(BasicBolt):
    def __init__(self):
        self.producer = cfg.config.kafka_producer
        super().__init__(script=__file__)

    def declareOutputFields(self):
        """Final bolt should not define any output fields
        you should always return empty array for the final bolt
        """
        return []

    def process(self, tup):
        # record is supposed to be dict with data we need to persist
        topic, record = tup.values
        log.debug("{}:{}".format(topic, record))

        # if record is not dictionary - bypass the record
        if isinstance(record, dict):
            try:
                persist = getattr(
                    records,
                    inflection.camelize("{}_record".format(topic)))
                persist(**record).save()
            except:               
                problem = {
                    "topic": topic,
                    "record": record,
                    "error": traceback.format_exc()
                }
                # send to *_failed kafka topic so we can reprocess later
                self.producer.send("{}_failed".format(topic), problem)
                pp = pprint.PrettyPrinter(indent=4)
                log.debug(pp.pformat(problem))


def run():
    """Default function needed to run bolt as task
    inside Apache Storm Topology
    """
    PersistenceBolt().run()