Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/test/MockApiProcessor.java (Java)

Code before:
public class MockApiProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, VIn, KOut, VOut> {

    private final ArrayList<Record<KIn, VIn>> processed = new ArrayList<>();
    private final Map<KIn, ValueAndTimestamp<VIn>> lastValueAndTimestampPerKey = new HashMap<>();

    private final ArrayList<Long> punctuatedStreamTime = new ArrayList<>();
    private final ArrayList<Long> punctuatedSystemTime = new ArrayList<>();

    private Cancellable scheduleCancellable;

    private final PunctuationType punctuationType;
    private final Instant startTime;
    private final long scheduleInterval;

    private boolean commitRequested = false;
    private ProcessorContext<KOut, VOut> context;

    public MockApiProcessor(final PunctuationType punctuationType,
                            final long scheduleInterval) {
        this(punctuationType, null, scheduleInterval);
    }

    public MockApiProcessor(final PunctuationType punctuationType,
                            final Instant startTime,
                            final long scheduleInterval) {
        this.punctuationType = punctuationType;
        this.startTime = startTime;
        this.scheduleInterval = scheduleInterval;
    }

    public MockApiProcessor() {
        this(PunctuationType.STREAM_TIME, -1);
    }

    @Override
    public void init(final ProcessorContext<KOut, VOut> context) {
        this.context = context;
        if (scheduleInterval > 0L) {
            scheduleCancellable = context.schedule(
                startTime,
                Duration.ofMillis(scheduleInterval),
                punctuationType,
                (punctuationType == PunctuationType.STREAM_TIME ? punctuatedStreamTime : punctuatedSystemTime)::add
            );
        }
    }

    @Override
    public void process(final Record<KIn, VIn> record) {
        final KIn key = record.key();
        final VIn value = record.value();

Review comment (suggestion):
Same consideration here to make the `startTime` an `Optional` or use `Instant.EPOCH_TIME` as a sentinel value when calling as line 58 above

Code after:
public class MockApiProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, VIn, KOut, VOut> {

    private final ArrayList<Record<KIn, VIn>> processed = new ArrayList<>();
    private final Map<KIn, ValueAndTimestamp<VIn>> lastValueAndTimestampPerKey = new HashMap<>();

    private final ArrayList<Long> punctuatedStreamTime = new ArrayList<>();
    private final ArrayList<Long> punctuatedSystemTime = new ArrayList<>();

    private Cancellable scheduleCancellable;

    private final PunctuationType punctuationType;
    private final Instant startTime;
    private final long scheduleInterval;

    private boolean commitRequested = false;
    private ProcessorContext<KOut, VOut> context;

    public MockApiProcessor(final PunctuationType punctuationType,
                            final long scheduleInterval) {
        this.punctuationType = punctuationType;
        this.startTime = null;  // unanchored schedule so start time is undefined
        this.scheduleInterval = scheduleInterval;
    }

    public MockApiProcessor(final PunctuationType punctuationType,
                            final Instant startTime,
                            final long scheduleInterval) {
        this.punctuationType = punctuationType;
        this.startTime = startTime;
        this.scheduleInterval = scheduleInterval;
    }

    public MockApiProcessor() {
        this(PunctuationType.STREAM_TIME, -1);
    }

    @Override
    public void init(final ProcessorContext<KOut, VOut> context) {
        this.context = context;
        if (scheduleInterval > 0L) {
            scheduleCancellable = (startTime == null)
                    ? context.schedule(
                    Duration.ofMillis(scheduleInterval),
                    punctuationType,
                    (punctuationType == PunctuationType.STREAM_TIME ? punctuatedStreamTime : punctuatedSystemTime)::add
            )
                    : context.schedule(
                    startTime,
                    Duration.ofMillis(scheduleInterval),
                    punctuationType,
                    (punctuationType == PunctuationType.STREAM_TIME ? punctuatedStreamTime : punctuatedSystemTime)::add