@Override
    public final void writeRecord(OUT record) throws IOException {
        checkAsyncErrors();
        tryAcquire(1);
        final CompletionStage<V> completionStage;
        try {
            completionStage = send(record);
        } catch (Throwable e) {
            semaphore.release();
            throw e;
        }
        completionStage.whenComplete(
                (result, throwable) -> {
                    if (throwable == null) {
                        callback.onSuccess(result);
                    } else {
                        callback.onFailure(throwable);
                    }
                });
    }
@Test
    void testMaxConcurrentRequestsReached() throws Exception {
        try (TestOutputFormat testOutputFormat =
                createOpenedTestOutputFormat(Duration.ofMillis(1))) {
            CompletableFuture<Void> completableFuture = new CompletableFuture<>();
            testOutputFormat.enqueueCompletableFuture(completableFuture);
            testOutputFormat.enqueueCompletableFuture(completableFuture);
            testOutputFormat.writeRecord("writeRecord #1");

            // writing a second time while the first request is still not completed and the
            // outputFormat is set for maxConcurrentRequests=1 will fail
            assertThatThrownBy(
                            () -> testOutputFormat.writeRecord("writeRecord #2"),
                            "Sending value should have experienced a TimeoutException.")
                    .hasCauseInstanceOf(TimeoutException.class);
            completableFuture.complete(null);
        }
    }