@Override
    public CompletableFuture<Acknowledge> sendEvent(OperatorEvent evt) {
        if (!isReady()) {
            throw new FlinkRuntimeException("SubtaskGateway is not ready, task not yet running.");
        }

        final SerializedValue<OperatorEvent> serializedEvent;
        try {
            serializedEvent = new SerializedValue<>(evt);
        } catch (IOException e) {
            // we do not expect that this exception is handled by the caller, so we make it
            // unchecked so that it can bubble up
            throw new FlinkRuntimeException("Cannot serialize operator event", e);
        }

        final Callable<CompletableFuture<Acknowledge>> sendAction =
                subtaskAccess.createEventSendAction(serializedEvent);

        final CompletableFuture<Acknowledge> sendResult = new CompletableFuture<>();
        final CompletableFuture<Acknowledge> result =
                sendResult.whenCompleteAsync(
                        (success, failure) -> {
                            if (failure != null && subtaskAccess.isStillRunning()) {
                                String msg =
                                        String.format(
                                                EVENT_LOSS_ERROR_MESSAGE,
                                                evt,
                                                subtaskAccess.subtaskName());
                                Runnables.assertNoException(
                                        () ->
                                                subtaskAccess.triggerTaskFailover(
                                                        new FlinkException(msg, failure)));
                            }
                        },
                        mainThreadExecutor);

        mainThreadExecutor.execute(
                () -> {
                    sendEventInternal(sendAction, sendResult);
                    incompleteFuturesTracker.trackFutureWhileIncomplete(result);
                });

        return result;
    }
@Test
    void eventsPassThroughOpenGateway() {
        final EventReceivingTasks receiver = EventReceivingTasks.createForRunningTasks();
        final SubtaskGatewayImpl gateway =
                new SubtaskGatewayImpl(
                        getUniqueElement(receiver.getAccessesForSubtask(11)),
                        ComponentMainThreadExecutorServiceAdapter.forMainThread(),
                        new IncompleteFuturesTracker());

        final OperatorEvent event = new TestOperatorEvent();
        final CompletableFuture<Acknowledge> future = gateway.sendEvent(event);

        assertThat(receiver.events).containsExactly(new EventWithSubtask(event, 11));
        assertThat(future).isDone();
    }