Process records performing retries as needed. Skip "poison pill" records.

@param records Data records to be processed.
978594
private void processRecordsWithRetries(List<Record> records) {
        for (Record record : records) {
            boolean processedSuccessfully = false;
            for (int i = 0; i < NUM_RETRIES; i++) {
                try {
                    //
                    // Logic to process record goes here.
                    //
                    processSingleRecord(record);

                    processedSuccessfully = true;
                    break;
                } catch (Throwable t) {
                    LOG.warn("Caught throwable while processing record " + record, t);
                }

                // backoff if we encounter an exception.
                try {
                    Thread.sleep(BACKOFF_TIME_IN_MILLIS);
                } catch (InterruptedException e) {
                    LOG.debug("Interrupted sleep", e);
                }
            }

            if (!processedSuccessfully) {
                LOG.error("Couldn't process record " + record + ". Skipping the record.");
            }
        }
    }
284494
def _delete_record(self, identifier=None, rtype=None, name=None, content=None):
        """"""
        records = self._raw_records(identifier, rtype, name, content)
        LOGGER.debug('delete_records: %s', [rec['id'] for rec in records])
        self._update_records(records, {
            'deleterecord': True,
            'type': rtype,
            'hostname': name,
            'destination': content,
        })
        LOGGER.debug('delete_record: %s', True)
        return True
966495
def handle(self, record):
        
        record = self.prepare(record)
        for handler in self.handlers:
            handler(record)
403557
function requeue(ddqBackendInstance, recordId, callback) {
        var record;

        record = ddqBackendInstance.getRecord(recordId);
        record.isProcessing = false;
        record.requeued = true;
        callback();
    }
698259
def processed_records(self, records):
        
        for record in records:
            self.logger.debug('Marking as processed: %s', record)
            record.processed = True
        db.session.commit()
862611
public void processEvent(LogRecord record) {
		byte[] bytes;

		SerializationObject serializationObject = pool.getSerializationObject();
		try {
			bytes = serializationObject.serialize(record);
		} finally {
			pool.returnSerializationObject(serializationObject);
		}

		synchronized(this) {
			if (traceWriter != null && record.getLevel().intValue() < traceThreshold) {
				traceWriter.logRecord(record.getMillis(), bytes);
			} else if (logWriter != null) {
				logWriter.logRecord(record.getMillis(), bytes);
			}
		}
	}
923207
public void updateRecord(final ORecordInternal<?> record) {
    if (isEnabled() && record.getIdentity().getClusterId() != excludedCluster && record.getIdentity().isValid()) {
      underlying.lock(record.getIdentity());
      try {
        if (underlying.get(record.getIdentity()) != record)
          underlying.put(record);
      } finally {
        underlying.unlock(record.getIdentity());
      }
    }

    secondary.updateRecord(record);
  }
1200300
def filter_records(self, records):
        
        for record in records:
            try:
                filtered = self.filter_record(record)
                assert (filtered)
                # Quick tracking whether the sequence was modified
                if filtered.seq == record.seq:
                    self.passed_unchanged += 1
                else:
                    self.passed_changed += 1
                yield filtered
            except FailedFilter as e:
                self.failed += 1
                v = e.value
                if self.listener:
                    self.listener(
                        'failed_filter',
                        record,
                        filter_name=self.name,
                        value=v)
1287488
public void doSet(Rec record) throws DBException
    {
        try   {
            m_objLastModBookmark = null;
            m_tableRemote.set(this.getDataSource(), record.getOpenMode());
        } catch (RemoteException ex) {
            throw new DBException(ex.getMessage());
        }
    }
783404
def setCheckedRecords( self, records ):
        
        QApplication.sendPostedEvents(self, -1)
        indexes = []
        
        for i in range(self.count()):
            record = self.recordAt(i)
            if record is not None and record in records:
                indexes.append(i)
        
        self.setCheckedIndexes(indexes)
1024125
public void setBillingRecords(java.util.Collection<BillingRecord> billingRecords) {
        if (billingRecords == null) {
            this.billingRecords = null;
            return;
        }

        this.billingRecords = new com.amazonaws.internal.SdkInternalList<BillingRecord>(billingRecords);
    }
808041
public int handleGetRawRecordData(Rec record)
    {
        int iErrorCode = Constant.NORMAL_RETURN;
        record = this.setNodeIndex(RESET_INDEX, record);   // Being careful
        int iNodeCount = 1;
        if (this.getNodeType() == MessageRecordDesc.NON_UNIQUE_NODE)
            iNodeCount = this.getMessage().getNodeCount(this.getFullKey(null));
        for (int index = 1; index <= iNodeCount; index++)
        {
            if (this.getNodeType() == MessageRecordDesc.NON_UNIQUE_NODE)
                record = this.setNodeIndex(index, record);
            
            iErrorCode = super.handleGetRawRecordData(record);
            
            if (iErrorCode != Constant.NORMAL_RETURN)
                break;
        }
        if (this.getNodeType() == MessageRecordDesc.NON_UNIQUE_NODE)
            this.setNodeIndex(END_OF_NODES, record);
        return iErrorCode;
    }
166038
public void check(Map<String,Object> record) throws BadDataException {
        if(Objects.isNullOrEmpty(record.get("id")))
            throw new BadDataException("id");
    }
717768
public void fixRecord(Record record)
    {
        super.fixRecord(record);
        if (this.getProperty("field") != null)
        {
            BaseField field = this.getMainRecord().getField(this.getProperty("field").toString());
            if (field != null)
                this.fixCapitalization(field);
        }
    }
282752
public void setHistoryRecords(java.util.Collection<HistoryRecordEntry> historyRecords) {
        if (historyRecords == null) {
            this.historyRecords = null;
            return;
        }

        this.historyRecords = new com.amazonaws.internal.SdkInternalList<HistoryRecordEntry>(historyRecords);
    }
1174350
public function onLocalRecordsLoaded($records, BaseRecords $baseRecords)
    {
        if (!$this->checkRecordPlugin($baseRecords)) {
            return;
        }

        if (count($records) > 0) {
            $this->updater->setLocalRecord($records[0]->getPlayer()->getNickname(), $records[0]->getCheckpoints());
        } else {
            $this->updater->setLocalRecord("-", []);
        }
    }
955663
public function onLocalRecordsFirstRecord(Record $record, $records, $position, BaseRecords $baseRecords)
    {
        if (!$this->checkRecordPlugin($baseRecords)) {
            return;
        }

        $this->updater->setLocalRecord($record->getPlayer()->getNickname(), $record->getCheckpoints());
    }