Repository: apache/flink
File: flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/lookup/keyordered/Epoch.java (Java)

Code before:

    private EpochStatus status;

    public Epoch(Watermark watermark) {
        this.status = EpochStatus.OPEN;
        this.ongoingRecordCount = 0;
        this.advanceWatermark = null;
        this.output = null;
        this.outputQueue = new LinkedList<>();
        this.watermark = watermark;
    }

    /** Add resultFuture to the output queue. */
    public void collect(StreamElementQueueEntry<OUT> resultFuture) {
        outputQueue.add(resultFuture);
    }

    public void setOutput(Consumer<StreamElementQueueEntry<OUT>> outputConsumer) {
        if (output == null) {
            this.output = outputConsumer;
        }
    }

    public void decrementCount() {
        ongoingRecordCount--;
        Preconditions.checkState(ongoingRecordCount >= 0);
    }

    public void incrementCount() {
        ongoingRecordCount++;
    }

    public Watermark getWatermark() {
        return watermark;
    }

    /**
     * Try to finish this epoch.
     *
     * @return whether this epoch has been finished.
     */
    boolean tryFinish() {
        if (this.status == EpochStatus.FINISHED) {
            return true;
        }
        while (!outputQueue.isEmpty()) {
            assert output != null;
            output.accept(outputQueue.poll());
        }
        if (ongoingRecordCount == 0 && this.status == EpochStatus.CLOSED) {
            this.status = EpochStatus.FINISHED;

Review comment (suggestion):
I am looking for the details of the inconsistency. I cannot find a description in the PR or the Jira. It seems odd to call decrementCount. Some thoughts: 
-I wonder why the caller cannot check that the decrementCount should not be called if there is nothing to decrement.   
- It would seem more fitting to do the precondition check before the decrement. As it is pre check.  Preconditions.checkState(ongoingRecordCount > 0);
- if this is to cope with multiple threads accessing the variable then the logic should be made thread safe. 

Code after:

    private EpochStatus status;

    public Epoch(Watermark watermark) {
        this.status = EpochStatus.OPEN;
        this.ongoingRecordCount = 0;
        this.advanceWatermark = null;
        this.output = null;
        this.outputQueue = new LinkedList<>();
        this.watermark = watermark;
    }

    /** Add resultFuture to the output queue. */
    public void collect(StreamElementQueueEntry<OUT> resultFuture) {
        outputQueue.add(resultFuture);
    }

    public void setOutput(Consumer<StreamElementQueueEntry<OUT>> outputConsumer) {
        if (output == null) {
            this.output = outputConsumer;
        }
    }

    public void decrementCount() {
        Preconditions.checkState(ongoingRecordCount > 0);
        ongoingRecordCount--;
    }

    public void incrementCount() {
        ongoingRecordCount++;
    }

    public Watermark getWatermark() {
        return watermark;
    }

    /**
     * Try to finish this epoch.
     *
     * @return whether this epoch has been finished.
     */
    boolean tryFinish() {
        if (this.status == EpochStatus.FINISHED) {
            return true;
        }
        while (!outputQueue.isEmpty()) {
            assert output != null;
            output.accept(outputQueue.poll());
        }
        if (ongoingRecordCount == 0 && this.status == EpochStatus.CLOSED) {
            this.status = EpochStatus.FINISHED;