@Override
    public void run(Runnable task) {
        long startNanos = System.nanoTime();

        boolean publishCurrentTask = publishCurrentTask();

        if (publishCurrentTask) {
            currentTask = task;
        }

        try {
            task.run();
        } finally {
            if (publishCurrentTask) {
                currentTask = null;
            }

            record(task, startNanos);
        }
    }
@Test
    public void runOperation_whenWaitingNeeded() {
        final AtomicLong counter = new AtomicLong();

        DummyWaitingOperation op = new DummyWaitingOperation() {
            @Override
            public void run() {
                counter.incrementAndGet();
            }
        };
        op.setPartitionId(operationRunner.getPartitionId());

        operationRunner.run(op);
        assertEquals(0, counter.get());
        // verify that the response handler was not called
        verify(responseHandler, never()).sendResponse(same(op), any());
    }