@Override
    public FsStateChangelogWriter createWriter(
            String operatorID, KeyGroupRange keyGroupRange, MailboxExecutor mailboxExecutor) {
        UUID logId = new UUID(0, logIdGenerator.getAndIncrement());
        LOG.info("createWriter for operator {}/{}: {}", operatorID, keyGroupRange, logId);
        return new FsStateChangelogWriter(
                logId,
                keyGroupRange,
                uploader,
                preEmptivePersistThresholdInBytes,
                mailboxExecutor,
                changelogRegistry,
                localRecoveryConfig,
                localChangelogRegistry);
    }
@Test
    public void testDeadlockOnUploadCompletion() throws Throwable {
        int capacity = 10; // in bytes, allow the first two uploads without waiting (see below)
        CountDownLatch remainingUploads = new CountDownLatch(3);
        BlockingUploader blockingUploader = new BlockingUploader();
        CompletableFuture<Void> unblockFuture = new CompletableFuture<>();
        new Thread(
                        () -> {
                            try {
                                remainingUploads.await();
                                blockingUploader.unblock();
                                unblockFuture.complete(null);
                            } catch (Throwable e) {
                                unblockFuture.completeExceptionally(e);
                            }
                        })
                .start();
        MailboxExecutorImpl mailboxExecutor =
                new MailboxExecutorImpl(
                        new TaskMailboxImpl(), 0, StreamTaskActionExecutor.IMMEDIATE);
        try (BatchingStateChangeUploadScheduler scheduler =
                        new BatchingStateChangeUploadScheduler(
                                0, // schedule immediately
                                0, // schedule immediately
                                RetryPolicy.NONE,
                                blockingUploader,
                                1,
                                capacity,
                                createUnregisteredChangelogStorageMetricGroup()) {
                            @Override
                            public void upload(UploadTask uploadTask) throws IOException {
                                remainingUploads.countDown();
                                super.upload(uploadTask);
                            }
                        };
                StateChangelogWriter<?> writer =
                        new FsStateChangelogStorage(
                                        scheduler,
                                        0,
                                        TaskChangelogRegistry.NO_OP, /* persist immediately */
                                        TestLocalRecoveryConfig.disabled())
                                .createWriter(
                                        new OperatorID().toString(),
                                        KeyGroupRange.of(0, 0),
                                        mailboxExecutor); ) {
            // 1. start with 1-byte request - releasing only it will NOT allow proceeding in 3, but
            // still involves completion callback, which can deadlock
            writer.append(0, new byte[1]);
            // 2. exceed capacity
            writer.append(0, new byte[capacity]);
            // 3. current thread will block until both previous requests are completed
            // verify that completion can proceed while this thread is waiting
            writer.append(0, new byte[1]);
        }
        // check unblocking thread exit status
        unblockFuture.join();
    }