public CompletableFuture<Long> compact(String topic) {
        return RawReader.create(pulsar, topic, COMPACTION_SUBSCRIPTION, false).thenComposeAsync(
                this::compactAndCloseReader, scheduler);
    }
@Test
    public void testCompactedWithConcurrentSend() throws Exception {
        String topic = "persistent://my-property/use/my-ns/testCompactedWithConcurrentSend";

        @Cleanup
        Producer<byte[]> producer = pulsarClient.newProducer().topic(topic)
                .enableBatching(false)
                .messageRoutingMode(MessageRoutingMode.SinglePartition)
                .create();

        var future = CompletableFuture.runAsync(() -> {
            for (int i = 0; i < 100; i++) {
                try {
                    producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i).getBytes()).send();
                } catch (PulsarClientException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).get();
        PulsarTopicCompactionService topicCompactionService =
                (PulsarTopicCompactionService) persistentTopic.getTopicCompactionService();

        Awaitility.await().untilAsserted(() -> {
            long compactedLedgerId = compact(topic);
            Thread.sleep(300);
            Optional<CompactedTopicContext> compactedTopicContext = topicCompactionService.getCompactedTopic()
                    .getCompactedTopicContext();
            Assert.assertTrue(compactedTopicContext.isPresent());
            Assert.assertEquals(compactedTopicContext.get().ledger.getId(), compactedLedgerId);
        });

        Position lastCompactedPosition = topicCompactionService.getLastCompactedPosition().get();
        Entry lastCompactedEntry = topicCompactionService.readLastCompactedEntry().get();

        Assert.assertTrue(PositionFactory.create(lastCompactedPosition.getLedgerId(), lastCompactedPosition.getEntryId())
                .compareTo(lastCompactedEntry.getLedgerId(), lastCompactedEntry.getEntryId()) >= 0);

        future.join();
    }