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

[dmitry-saritasa]

hi Barry,

Example of the bolt on my side. When I run topology - everything works normal and my messages are persisted into database, but Storm UI shows NO acknowledged tuples. Zero. Although my bolt is inherited from BasicBolt that's supposed to do automated ack().

Thoughts?

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()

[barrywhart]

The ack() happens here. You might try adding logging to that function to make sure it's being executed. Or temporarily switch to Bolt and see if it helps to ack() explicitly.