Java 类com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput 实例源码

@Override
@SneakyThrows
public void shutdown(ShutdownInput shutdownInput) {
    IRecordProcessorCheckpointer checkpointer = shutdownInput.getCheckpointer();
    ShutdownReason reason = shutdownInput.getShutdownReason();
    log.info("Finalizado trabajo: {}.", reason);
    checkpointer.checkpoint();
}

@Override
public void shutdown(ShutdownInput input) {

    Thread.currentThread().setName(kinesisShardId);
    if (input.getShutdownReason() == ShutdownReason.ZOMBIE) {
        /* This happens because we have lost our lease. Either because of re-balancing
        * (there is another NomNom running on another host), or because we have failed
        * failed to renew our leases, which would happen if we are too busy.
        *
        * It happens when a new version of NomNom is deployed, since the two versions
        * will run side by side for a few seconds before the old one is terminated. And
        * the new one will try to take a few leases at startup.
        */
        LOG.warn("We're a ZOMBIE - someone stole our lease. Quitting.");
    } else {
        /* This happens when a shard is split or merged, meaning that it stops existing
         * and we have other shards to process instead. Very rare.
         */
        LOG.warn("Shard is shutting down, reason: {}", input.getShutdownReason());
        try {
            input.getCheckpointer().checkpoint();
        } catch (Exception e) {
            LOG.error("Failed to checkpoint after shard shutdown", e);
        }
    }
    LOG.info("");
}

/**
 * Invoked by the Amazon Kinesis Client Library to indicate it will no longer send data records to this
 * RecordProcessor instance.
 *
 * @param shutdownInput Provides information and capabilities (eg checkpointing) related to shutdown of this record
 *        processor.
 */
@Override
public void shutdown(ShutdownInput shutdownInput) {
  LOG.info("Shutting down {} with reason:{}", this, shutdownInput.getShutdownReason());

  Validate.isTrue(!shutdownRequested, String.format("KCL called shutdown more than once for processor %s.", this));
  shutdownRequested = true;
  // shutdown reason TERMINATE indicates that the shard is closed due to re-sharding. It also indicates that all the
  // records from the shard have been delivered to the consumer and the consumer is expected to checkpoint the
  // progress.
  if (shutdownInput.getShutdownReason() == ShutdownReason.TERMINATE) {
    // We need to ensure that all records are processed and checkpointed before going ahead and marking the processing
    // complete by calling checkpoint() on KCL. We need to checkpoint the completion state for parent shard, for KCL
    // to consume from the child shard(s).
    try {
      LOG.info("Waiting for all the records for {} to be processed.", this);
      // Let's poll periodically and block until the last processed record is checkpointed. Also handle the case
      // where there are no records received for the processor, in which case the lastProcessedRecordSeqNumber will
      // be null.
      while (lastProcessedRecordSeqNumber != null
          && !lastProcessedRecordSeqNumber.equals(lastCheckpointedRecordSeqNumber)) {
        Thread.sleep(POLL_INTERVAL_DURING_PARENT_SHARD_SHUTDOWN_MS);
      }
      LOG.info("Final checkpoint for {} before shutting down.", this);
      shutdownInput.getCheckpointer().checkpoint();
    } catch (Exception e) {
      LOG.warn("An error occurred while committing the final checkpoint in the parent shard {}", this, e);
    }
  }
  listener.onShutdown(ssp);
}

static void shutDownProcessor(KinesisRecordProcessor processor, ShutdownReason reason) {
  try {
    ShutdownInput shutdownInput = Mockito.mock(ShutdownInput.class);
    when(shutdownInput.getShutdownReason()).thenReturn(reason);
    when(shutdownInput.getCheckpointer()).thenReturn(getCheckpointer(processor));
    processor.shutdown(shutdownInput);
  } catch (Exception ex) {
    throw new RuntimeException(ex);
  }
}

/**
 * We don't checkpoint on SHUTDOWN_REQUESTED because we currently always
 * checkpoint each batch in {@link #processRecords}.
 *
 * @param shutdownInput {@inheritDoc}
 */
@Override
public void shutdown(ShutdownInput shutdownInput) {
  LOG.info("Shutting down record processor for shard: {}", shardId);

  if (ShutdownReason.TERMINATE.equals(shutdownInput.getShutdownReason())) {
    // Shard is closed / finished processing. Checkpoint all processing up to here.
    try {
      shutdownInput.getCheckpointer().checkpoint();
      LOG.debug("Checkpointed due to record processor shutdown request.");
    } catch (InvalidStateException | ShutdownException e) {
      LOG.error("Error checkpointing batch: {}", e.toString(), e);
    }
  }
}

@Override
public void shutdown(ShutdownInput shutdownInput) {
}