Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java (Java)

Code before:
        currentMaxDesiredWatermark = event.getMaxWatermark();
        sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
    }

    @Override
    public void updateIdle(boolean isIdle) {
        this.idle = isIdle;
    }

    @Override
    public void updateCurrentEffectiveWatermark(long watermark) {
        sampledLatestWatermark.addLatest(watermark);
        checkWatermarkAlignment();
    }

    @Override
    public void updateCurrentSplitWatermark(String splitId, long watermark) {
        WatermarkSampler splitWatermarkSampler = checkNotNull(sampledSplitWatermarks.get(splitId));
        splitWatermarkSampler.addLatest(watermark);
        if (!currentlyIdleSplits.contains(splitId)) {
            maybePauseSplit(splitId);
        }
    }

    private void maybePauseSplit(String splitId) {
        WatermarkSampler splitWatermarkSampler = checkNotNull(sampledSplitWatermarks.get(splitId));
        long oldestSampledWatermark = splitWatermarkSampler.getOldestSample();
        // oldestSampledWatermark can be only updated after adding new latest if sampling capacity
        // is 0, but we still need to handle that
        if (oldestSampledWatermark > currentMaxDesiredWatermark
                && !currentlyPausedSplits.contains(splitId)) {
            pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
            currentlyPausedSplits.add(splitId);
        }
    }

    @Override
    public void updateCurrentSplitIdle(String splitId, boolean idle) {
        final InternalSourceSplitMetricGroup splitMetricGroup =
                this.getOrCreateSplitMetricGroup(splitId);
        if (idle == currentlyIdleSplits.contains(splitId)) {
            return;
        }
        if (idle) {
            LOG.info("[{}] Marking split idle", splitId);
            currentlyIdleSplits.add(splitId);
            splitMetricGroup.markIdle();
        } else {
            LOG.info("[{}] Marking split not idle", splitId);
            currentlyIdleSplits.remove(splitId);
            splitMetricGroup.markNotIdle();

Review comment (suggestion):
the method name is strange - I was expecting  it to start with a verb like the other methods. Maybe `updateCurrentSplitPausedWatermark 

Code after:
    private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
        currentMaxDesiredWatermark = event.getMaxWatermark();
        sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
    }

    @Override
    public void updateIdle(boolean isIdle) {
        this.idle = isIdle;
    }

    @Override
    public void updateCurrentEffectiveWatermark(long watermark) {
        sampledLatestWatermark.addLatest(watermark);
        checkWatermarkAlignment();
    }

    @Override
    public void updateCurrentSplitWatermark(String splitId, long watermark) {
        WatermarkSampler splitWatermarkSampler = checkNotNull(sampledSplitWatermarks.get(splitId));
        splitWatermarkSampler.addLatest(watermark);
        if (!currentlyIdleSplits.contains(splitId)) {
            maybePauseSplit(splitId);
        }
    }

    private void maybePauseSplit(String splitId) {
        WatermarkSampler splitWatermarkSampler = checkNotNull(sampledSplitWatermarks.get(splitId));
        long oldestSampledWatermark = splitWatermarkSampler.getOldestSample();
        // oldestSampledWatermark can be only updated after adding new latest if sampling capacity
        // is 0, but we still need to handle that
        if (oldestSampledWatermark > currentMaxDesiredWatermark
                && !currentlyPausedSplits.contains(splitId)) {
            pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
            currentlyPausedSplits.add(splitId);
        }
    }

    @Override
    public void updateCurrentSplitIdle(String splitId, boolean idle) {
        final InternalSourceSplitMetricGroup splitMetricGroup =
                this.getOrCreateSplitMetricGroup(splitId);
        if (idle == currentlyIdleSplits.contains(splitId)) {
            return;
        }
        if (idle) {
            LOG.info("[{}] Marking split idle", splitId);
            currentlyIdleSplits.add(splitId);
            splitMetricGroup.markIdle();
        } else {
            LOG.info("[{}] Marking split not idle", splitId);
            currentlyIdleSplits.remove(splitId);