@Override
    public int size() {
        return eventHandler.size();
    }
@Test
    public void testSize() throws Exception {
        try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testEmpty")) {
            assertTrue(queue.isEmpty());
            CompletableFuture<Void> future = new CompletableFuture<>();
            queue.append(() -> future.get());
            assertFalse(queue.isEmpty());
            assertEquals(1, queue.size());
            queue.append(() -> future.get());
            assertEquals(2, queue.size());
            future.complete(null);
            TestUtils.waitForCondition(() -> queue.isEmpty(), "Failed to see the queue become empty.");
            queue.scheduleDeferred("later",
                    __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + HOURS.toNanos(1)),
                    () -> {
                    });
            assertFalse(queue.isEmpty());
            queue.scheduleDeferred("soon",
                    __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + TimeUnit.MILLISECONDS.toNanos(1)),
                    () -> {
                    });
            assertFalse(queue.isEmpty());
            queue.cancelDeferred("later");
            queue.cancelDeferred("soon");
            TestUtils.waitForCondition(() -> queue.isEmpty(), "Failed to see the queue become empty.");
            assertTrue(queue.isEmpty());
        }
    }