public TaskAcknowledgeResult acknowledgeCoordinatorState(
            OperatorInfo coordinatorInfo, @Nullable ByteStreamStateHandle stateHandle) {

        synchronized (lock) {
            if (disposed) {
                return TaskAcknowledgeResult.DISCARDED;
            }

            final OperatorID operatorId = coordinatorInfo.operatorId();
            OperatorState operatorState = operatorStates.get(operatorId);

            // sanity check for better error reporting
            if (!notYetAcknowledgedOperatorCoordinators.remove(operatorId)) {
                return operatorState != null && operatorState.getCoordinatorState() != null
                        ? TaskAcknowledgeResult.DUPLICATE
                        : TaskAcknowledgeResult.UNKNOWN;
            }

            if (operatorState == null) {
                operatorState =
                        new OperatorState(
                                operatorId,
                                coordinatorInfo.currentParallelism(),
                                coordinatorInfo.maxParallelism());
                operatorStates.put(operatorId, operatorState);
            }
            if (stateHandle != null) {
                operatorState.setCoordinatorState(stateHandle);
            }

            return TaskAcknowledgeResult.SUCCESS;
        }
    }
@Test
    void testDuplicateAcknowledgeCoordinator() throws Exception {
        final OperatorInfo coordinator = new TestingOperatorInfo();
        final PendingCheckpoint checkpoint = createPendingCheckpointWithCoordinators(coordinator);

        checkpoint.acknowledgeCoordinatorState(coordinator, new TestingStreamStateHandle());
        final TaskAcknowledgeResult secondAck =
                checkpoint.acknowledgeCoordinatorState(coordinator, null);

        assertThat(TaskAcknowledgeResult.DUPLICATE).isEqualTo(secondAck);
    }