@param array $chunk

@throws KinesisFailedRecordsException
When the Kinesis Result FailedRecordCount is grater than zero.
343665
protected function putRecords(array $chunk)
    {

        $args = ['StreamName' => $this->getStreamName()];

        foreach ($chunk as $message) {
            $args['Records'][] = [
                'Data' => $this->getMessageFactory()->serialize($message),
                // The partition key is the robo id so that from an individual robot
                // messages are pushed into same shard.
                'PartitionKey' => $message->getRoboId(),
            ];
        }

        if (!empty($args['Records'])) {
            /** @var Result $result */
            $result = $this->getClient()->putRecords($args);
            if ($result->get('FailedRecordCount') > 0) {
                throw new KinesisFailedRecordsException('Failed to push [' . $result->get('FailedRecordCount') . '] records');
            }
        }
    }
670082
public function flush(?string $chunk, bool $flushEmpty = false): void
	{
		if($this->isCGI)
		{
			if(!empty($chunk))
			{
				echo $chunk;

				flush();
			}
		}
		else
		{
			if(!empty($chunk) || $flushEmpty === true)
			{
				printf("%x\r\n%s\r\n", strlen($chunk), $chunk);

				flush();
			}
		}
	}
882578
private function validateArrayImplementation(
        array $input,
        CaptureAggregatorInterface $captures,
        ErrorAggregatorInterface $errors
    ): void {
        list($indexMap, $serialized) = $this->getSerializedRules();

        $blocks = BlockSerializer::unserializeBlocks($serialized);

        // the method is expected to be implemented by a class that uses this trait
        $context = $this->createContextStorageFromBlocks($blocks);

        BlockInterpreter::executeStarts(
            BlockSerializer::unserializeBlocksWithStart($serialized),
            $blocks,
            $context,
            $errors
        );
        foreach ($input as $key => $value) {
            $blockIndex = $indexMap[$key];
            BlockInterpreter::executeBlock($value, $blockIndex, $blocks, $context, $captures, $errors);
        }
        BlockInterpreter::executeEnds(
            BlockSerializer::unserializeBlocksWithEnd($serialized),
            $blocks,
            $context,
            $errors
        );
    }
891036
public function sendChunk($req, $chunk)
    {
        return $this->server->send($req->fd,
            "\x01" // protocol version
            . "\x06" // record type (STDOUT)
            . pack('nn', $req->id, strlen($chunk)) // id, content length
            . "\x00" // padding length
            . "\x00" // reserved
        ) && $this->server->send($req->fd, $chunk); // content
    }
371125
private void invokeChunk() {
        logger.entering(sourceClass, "invokeChunk");

        List<Object> chunkToWrite = new ArrayList<Object>();

        try {
            getTransactionManager().begin();
            this.openReaderAndWriter();
            getTransactionManager().commit();

            for (int currentChunk = 1; true; currentChunk++) {

                // Done with the previous chunk status so advance reference to next one.
                currentChunkStatus = getNextChunkStatusBasedOnPrevious();

                // Sequence surrounding beginCheckpoint() updated per MR
                // https://java.net/bugzilla/show_bug.cgi?id=5873
                setNextChunkTransactionTimeout();

                // Remember we "wrap" the built-in item-count + time-limit "algorithm"
                // in a CheckpointAlgorithm for ease in keeping the sequence consistent
                checkpointManager.beginCheckpoint();

                getTransactionManager().begin();

                for (ChunkListenerProxy chunkProxy : chunkListeners) {
                    chunkProxy.beforeChunk();
                }

                JoblogUtil.logToJobLogAndTraceOnly(Level.FINEST, "chunk.started", new Object[] {
                                                                                                 runtimeStepExecution.getStepName(), getJobInstanceId(), getJobExecutionId(),
                                                                                                 runtimeStepExecution.getMetrics() },
                                                   logger);

                chunkToWrite = readAndProcess();

                if (currentChunkStatus.wasMarkedForRollbackWithRetry()) {
                    rollbackAfterRetryableException(currentChunk);
                    continue;
                }

                // MR 1.0 Rev A clarified we'd only write a chunk with at least one item.
                // See, e.g. Sec 11.6 of Spec
                if (chunkToWrite.size() > 0) {
                    writeChunk(chunkToWrite);
                }

                if (currentChunkStatus.wasMarkedForRollbackWithRetry()) {
                    rollbackAfterRetryableException(currentChunk);
                    continue;
                }

                for (ChunkListenerProxy chunkProxy : chunkListeners) {
                    chunkProxy.afterChunk();
                }

                checkpointManager.checkpoint();

                updateNormalMetrics(chunkToWrite.size());

                runtimeStepExecution.setLastUpdatedTime(new Date());
                updateStepExecution();
                getPersistenceManagerService().updateStepThreadInstanceWithCheckpointData(getStepThreadInstance());

                getTransactionManager().commit();

                JoblogUtil.logToJobLogAndTraceOnly(Level.FINEST, "chunk.ended", new Object[] {
                                                                                               runtimeStepExecution.getStepName(), getJobInstanceId(), getJobExecutionId(),
                                                                                               runtimeStepExecution.getMetrics() },
                                                   logger);

                checkpointManager.endCheckpoint();

                // Send the step execution id as well
                publishCheckpointEvent(runtimeStepExecution.getStepName(), getJobInstanceId(), getJobExecutionId(), runtimeStepExecution.getTopLevelStepExecutionId());

                invokeCollectorIfPresent();

                // exit loop when last record is written or if we're stopping
                if (currentChunkStatus.hasReadNull() || currentChunkStatus.isStopping()) {
                    getTransactionManager().begin();

                    writerProxy.close();
                    readerProxy.close();

                    getTransactionManager().commit();
                    break;
                }
            }
        } catch (Throwable t) {
            // Note we've already carefully handled skippable and retryable exceptions.  Anything surfacing to this
            // level does not need to be considered as either.
            try {
                JoblogUtil.logToJobLogAndTraceOnly(Level.FINE, "chunk.rollback", new Object[] {
                                                                                                runtimeStepExecution.getStepName(), getJobInstanceId(), getJobExecutionId(),
                                                                                                runtimeStepExecution.getMetrics() },
                                                   logger);

                callReaderAndWriterCloseOnThrowable(t);

                // Signature is onError(Exception) so only try to call if we have an Exception, but not an Error.
                if (t instanceof Exception) {
                    callChunkListenerOnError((Exception) t);
                }

                // Let's not count only retry rollbacks but also non-retry rollbacks.
                runtimeStepExecution.getMetric(MetricImpl.MetricType.ROLLBACK_COUNT).incValue();

            } finally {
                getTransactionManager().rollback();
            }
            logger.exiting(sourceClass, "invokeChunk");
            throw new BatchContainerRuntimeException("Failure in Read-Process-Write Loop", t);
        }

        logger.finest("Exiting normally");
        logger.exiting(sourceClass, "invokeChunk");
    }
323778
public function sendChunk($req, $chunk)
    {
        $packet = "\x01" // protocol version
            . "\x06" // record type (STDOUT)
            . pack(
                'nn',
                $req->attrs->id,
                mb_orig_strlen($chunk)
            ) // id, content length
            . "\x00" // padding length
            . "\x00";// reserved

        return $this->write($packet) && $this->write($chunk); // content
    }
1231712
private void invokeChunk() {
		logger.entering(sourceClass, "invokeChunk");

		List<Object> chunkToWrite = new ArrayList<Object>();

		try {
			transactionManager.begin();
			this.openReaderAndWriter();
			transactionManager.commit();

			while (true) {

				// Done with the previous chunk status so advance reference to next one.
				currentChunkStatus = getNextChunkStatusBasedOnPrevious();

				// Sequence surrounding beginCheckpoint() updated per MR
				// https://java.net/bugzilla/show_bug.cgi?id=5873
				setNextChunkTransactionTimeout();

				// Remember we "wrap" the built-in item-count + time-limit "algorithm"
				// in a CheckpointAlgorithm for ease in keeping the sequence consistent
				checkpointManager.beginCheckpoint();

				transactionManager.begin();

				for (ChunkListenerProxy chunkProxy : chunkListeners) {
					chunkProxy.beforeChunk();
				}

				chunkToWrite = readAndProcess();

				if (currentChunkStatus.wasMarkedForRollbackWithRetry()) {
					rollbackAfterRetryableException();
					continue;
				}

				// MR 1.0 Rev A clarified we'd only write a chunk with at least one item.
				// See, e.g. Sec 11.6 of Spec
				if (chunkToWrite.size() > 0) {
					writeChunk(chunkToWrite);
				}

				if (currentChunkStatus.wasMarkedForRollbackWithRetry()) {
					rollbackAfterRetryableException();
					continue;
				}

				for (ChunkListenerProxy chunkProxy : chunkListeners) {
					chunkProxy.afterChunk();
				}

				checkpointManager.checkpoint();

				this.persistUserData();

				transactionManager.commit();

				checkpointManager.endCheckpoint();

				invokeCollectorIfPresent();

				updateNormalMetrics(chunkToWrite.size());

				// exit loop when last record is written or if we're stopping
				if (currentChunkStatus.hasReadNull() || currentChunkStatus.isStopping()) {
					transactionManager.begin();

					writerProxy.close();
					readerProxy.close();

					transactionManager.commit();
					break;
				}
			}
		} catch (Throwable t) {		
			// Note we've already carefully handled skippable and retryable exceptions.  Anything surfacing to this
			// level does not need to be considered as either.
			try {
				logger.log(Level.SEVERE, "Failure in Read-Process-Write Loop", t);

				callReaderAndWriterCloseOnThrowable(t);

				// Signature is onError(Exception) so only try to call if we have an Exception, but not an Error.
				if (t instanceof Exception) {
					callChunkListenerOnError((Exception)t);
				}
				// Let's not count only retry rollbacks but also non-retry rollbacks.
				stepContext.getMetric(MetricImpl.MetricType.ROLLBACK_COUNT).incValue();
			} finally {
				transactionManager.rollback();
			}
			logger.exiting(sourceClass, "invokeChunk");
			throw new BatchContainerRuntimeException("Failure in Read-Process-Write Loop", t);
		} 

		logger.finest("Exiting normally");
		logger.exiting(sourceClass, "invokeChunk");
	}
541398
protected function freezeArray(array &$array, array &$objects)
    {
        foreach ($array as &$value) {
            if (is_array($value)) {
                $this->freezeArray($value, $objects);
            } elseif (is_object($value)) {
                list($id,) = array_values($this->freeze($value, $objects));
                $value = '__freezer_' . $id;
            }
        }
    }
975866
public function failed(array $data)
	{
		$handler = $this->dispatcher->resolveHandler($command = unserialize($data['command']));

		if (method_exists($handler, 'failed'))
		{
			call_user_func([$handler, 'failed'], $command);
		}
	}
238734
private void onFlushBatch(Map<String, BulkWriteOperation> bulkWriteOperationMap)
    {
        if (!bulkWriteOperationMap.isEmpty())
        {
            for (BulkWriteOperation builder : bulkWriteOperationMap.values())
            {
                try
                {
                    builder.execute(getWriteConcern());
                }
                catch (BulkWriteException bwex)
                {
                    log.error("Batch insertion is not performed due to error in write command. Caused By: ", bwex);
                    throw new KunderaException(
                            "Batch insertion is not performed due to error in write command. Caused By: ", bwex);
                }
                catch (MongoException mex)
                {
                    log.error("Batch insertion is not performed. Caused By: ", mex);
                    throw new KunderaException("Batch insertion is not performed. Caused By: ", mex);
                }
            }
        }
    }
1256973
private function transformChunk(string $chunk)
    {
        if (null === $this->chunkTransformer) {
            return $chunk;
        }

        return \call_user_func($this->chunkTransformer, $chunk);
    }
1274950
public function chunkById($count, callable $callback, $column = null, $alias = null)
    {
        $column = is_null($column) ? $this->getModel()->getKeyName() : $column;

        $alias = is_null($alias) ? $column : $alias;

        $lastId = 0;

        do {
            $clone = clone $this;

            // We'll execute the query for the given page and get the results. If there are
            // no results we can just break and return from here. When there are results
            // we will call the callback with the current chunk of these results here.
            $results = $clone->forPageAfterId($count, $lastId, $column)->get();

            $countResults = $results->count();

            if ($countResults === 0) {
                break;
            }

            // On each chunk result set, we will pass them to the callback and then let the
            // developer take care of everything within the callback, which allows us to
            // keep the memory low for spinning through large result sets for working.
            if ($callback($results) === false) {
                return false;
            }

            $lastId = $results->last()->{$alias};
        } while ($countResults === $count);

        return true;
    }
984922
private function validateNewRow(CellChunk $chunk)
    {
        $this->isError(
            $this->row,
            'A new row cannot have existing state.'
        );
        $this->isError(
            !$chunk->getRowKey(),
            'A row key must be set.'
        );
        $this->isError(
            $chunk->getResetRow(),
            'A new row cannot be reset.'
        );
        $this->isError(
            $this->prevRowKey &&
            $this->prevRowKey === $chunk->getRowKey(),
            'A commit happened but the same key followed.'
        );
        $this->isError(!$chunk->getFamilyName(), 'A family must be set.');
        $this->isError(!$chunk->getQualifier(), 'A column qualifier must be set.');
        $this->validateValueSizeAndCommitRow($chunk);
    }
893436
public function executeSql($chunked = false)
    {
        try {
            list($sql, $params) = $this->asSql($this, $this->connection);
            $stmt = $this->connection->prepare($sql);
            foreach ($params as $index => $value) {
                // Columns/Parameters are 1-based,
                // so need to start at 1 instead of zero
                ++$index;
                $stmt->bindValue($index, $value);
            }

            $stmt->execute();

            return $stmt;
        } catch (\Exception $e) {
            throw QueryException::fromThrowable($e);
        }
    }
1048463
public function chunk($amount, callable $fn = null)
    {
        $this->prepare();

        $chunk_value = array_chunk($this->_map, $amount);
        $chunks = [];

        if (!is_null($fn) && is_callable($fn)) {
            foreach ($chunk_value as $chunk) {
                $return = $fn($chunk);
                if (!is_null($return)) {
                    $chunks[] = $return;
                }
            }
            return count($chunks) > 0 ? $chunks : null;
        }

        return $chunk_value;
    }
336953
protected function insert_chunk($table, array $chunk, array $columns) {
        $fieldssql = '('.implode(',', array_keys($columns)).')';

        $valuessql = '('.implode(',', array_fill(0, count($columns), '?')).')';
        $valuessql = implode(',', array_fill(0, count($chunk), $valuessql));

        $params = array();
        foreach ($chunk as $dataobject) {
            foreach ($columns as $field => $column) {
                $params[] = $this->normalise_value($column, $dataobject[$field]);
            }
        }

        $sql = "INSERT INTO {$this->prefix}$table $fieldssql VALUES $valuessql";

        list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
        $rawsql = $this->emulate_bound_params($sql, $params);

        $this->query_start($sql, $params, SQL_QUERY_INSERT);
        $result = $this->mysqli->query($rawsql);
        $this->query_end($result);
    }
1265544
public function chunk($count, $callback, $column = null, $order = 'asc')
    {
        $options = $this->getOptions();
        $column  = $column ?: $this->getPk($options);

        if (isset($options['order'])) {
            if ($this->config['debug']) {
                throw new DbException('chunk not support call order');
            }
            unset($options['order']);
        }

        $bind = $this->bind;

        if (is_array($column)) {
            $times = 1;
            $query = $this->options($options)->page($times, $count);
        } else {
            $query = $this->options($options)->limit($count);

            if (strpos($column, '.')) {
                list($alias, $key) = explode('.', $column);
            } else {
                $key = $column;
            }
        }

        $resultSet = $query->order($column, $order)->select();

        while (count($resultSet) > 0) {
            if ($resultSet instanceof Collection) {
                $resultSet = $resultSet->all();
            }

            if (false === call_user_func($callback, $resultSet)) {
                return false;
            }

            if (isset($times)) {
                $times++;
                $query = $this->options($options)->page($times, $count);
            } else {
                $end    = end($resultSet);
                $lastId = is_array($end) ? $end[$key] : $end->getData($key);

                $query = $this->options($options)
                    ->limit($count)
                    ->where($column, 'asc' == strtolower($order) ? '>' : '<', $lastId);
            }

            $resultSet = $query->bind($bind)->order($column, $order)->select();
        }

        return true;
    }