asyncbolt.scala

case class Completion(tuple:Tuple, exOpt: Option[Throwable] = None)

class SomeBolt extends BaseRichBolt {
  var completions = new LinkedBlockingQueue[Completion]()

  def flushCompletions = {
    if (completions.size() > 0) {
      val completionArr = new util.ArrayList[Completion]()
        completions.drainTo(completionArr)
        logger.debug("completing " + completionArr.size() + " tuples")
        completionArr.foreach { completion =>
          if (completion.exOpt.isEmpty) { 
            collector.ack(completion.tuple)
          } else {
            collector.reportError(completion.exOpt.get)
            collector.fail(completion.tuple)
          }
        }
    }
  }

  override def execute(tuple: Tuple): Unit = {
      flushCompletions
    ...
  }
  
  ...
}