@Override
    public void setIsProcessingBacklog(boolean isProcessingBacklog) {
        CheckpointCoordinator checkpointCoordinator =
                getCoordinatorContext().getCheckpointCoordinator();
        OperatorID operatorID = getCoordinatorContext().getOperatorId();
        if (checkpointCoordinator != null) {
            checkpointCoordinator.setIsProcessingBacklog(operatorID, isProcessingBacklog);
        }
        backlog = TernaryBoolean.fromBoolean(isProcessingBacklog);
        callInCoordinatorThread(
                () -> {
                    final IsProcessingBacklogEvent isProcessingBacklogEvent =
                            new IsProcessingBacklogEvent(isProcessingBacklog);
                    for (int i = 0; i < getCoordinatorContext().currentParallelism(); i++) {
                        sendEventToSourceOperatorIfTaskReady(i, isProcessingBacklogEvent);
                    }
                    return null;
                },
                "Failed to send BacklogEvent to reader.");
    }
@Test
    void testSetIsProcessingBacklog() throws Exception {
        sourceReady();
        registerReader(0, 0);
        context.setIsProcessingBacklog(true);

        for (int i = 0; i < context.currentParallelism(); ++i) {
            final List<OperatorEvent> events = receivingTasks.getSentEventsForSubtask(i);
            assertThat(events.get(events.size() - 1)).isEqualTo(new IsProcessingBacklogEvent(true));
        }

        registerReader(1, 0);
        context.setIsProcessingBacklog(false);
        registerReader(2, 0);

        for (int i = 0; i < context.currentParallelism(); ++i) {
            final List<OperatorEvent> events = receivingTasks.getSentEventsForSubtask(i);
            assertThat(events.get(events.size() - 1))
                    .isEqualTo(new IsProcessingBacklogEvent(false));
        }
    }