docs: 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");
      }