@Override
  public IncomingMessageEnvelope next() {
    IncomingMessageEnvelope envelope = super.next();

    // if trimming changelog is enabled, then switch to trim mode if we've consumed past the restore offset
    // (i.e., restoreOffset was null or current offset is > restoreOffset)
    if (this.trimEnabled && (restoreOffset == null || admin.offsetComparator(envelope.getOffset(), restoreOffset) > 0)) {
      mode = Mode.TRIM;
    }

    return envelope;
  }
Iterates over messages in the provided changelog {@link SystemStreamPartition} using the provided
{@link SystemConsumer} until all messages have been consumed.

The iterator has a {@link Mode} that depends on its position in the changelog SSP. If trim mode
is enabled, the mode switches to {@code TRIM} if the current message offset is greater than the
provided {@code restoreOffset}, or if {@code restoreOffset} is null.

The iterator mode is used during transactional state restore to determine which changelog SSP entries
should be restored and which ones need to be reverted / trimmed from the changelog topic.
samza-api/src/main/java/org/apache/samza/system/ChangelogSSPIterator.java