Method AsyncReadAck

src/observer/executor/scanner_impl.cc:499–538

}

void ScannerImpl::AsyncReadAck(std::shared_ptr<NotifyCell> notify_cell) {
  VLOG(12) << "[time] Check ACK start. [cf:qu] " << notify_cell->observed_column.family
           << notify_cell->observed_column.qualifier;

  const std::string& ack_qualifier_prefix = GetAckQualifierPrefix(
      notify_cell->observed_column.family, notify_cell->observed_column.qualifier);

  // use transaction to read column Ack
  std::shared_ptr<tera::Transaction> row_transaction(
      notify_cell->table->StartRowTransaction(notify_cell->row));
  NotificationContext* context = new NotificationContext();

  // read Acks
  tera::RowReader* row_reader = notify_cell->table->NewRowReader(notify_cell->row);

  const std::string& ack_qualifier =
      GetAckQualifier(ack_qualifier_prefix, notify_cell->observer->GetObserverName());
  context->ack_qualifier = ack_qualifier;

  row_reader->AddColumn(notify_cell->observed_column.family, ack_qualifier);

  context->notify_cell = notify_cell;
  context->scanner_impl = this;
  context->ack_transaction = row_transaction;
  row_reader->SetContext(context);
  row_reader->SetCallBack([](RowReader* ack_reader) {
    NotificationContext* context = (NotificationContext*)(ack_reader->GetContext());
    if (!context->scanner_impl->quit_) {
      context->scanner_impl->transaction_callback_threads_->AddTask(
          std::bind(&ScannerImpl::ValidateAckConfict, context->scanner_impl, ack_reader));
    } else {
      // call auto unlocker
      delete context;
      context = NULL;
      delete ack_reader;
    }
  });

  row_transaction->Get(row_reader);
}

std::string ScannerImpl::GetAckQualifierPrefix(const std::string& family,
                                               const std::string& qualifier) const {