@Override
    public Runnable get() {
        if (currentRunnable == null && currentStep != null) {
            currentRunnable = createRunnable(currentStep, state);
        }
        return currentRunnable;
    }
@Test
    public void step_supplier_ends_with_call_timeout_response_when_operation_timed_out() {
        // create node with force offload
        Config config = getConfig();
        config.setProperty(MapServiceContext.FORCE_OFFLOAD_ALL_OPERATIONS.getName(), "true");
        HazelcastInstance node = createHazelcastInstance(config);

        // create state to use for test verification
        AtomicReference<Object> expectedResponse = new AtomicReference<>();
        CountDownLatch latch = new CountDownLatch(1);

        // create and call slow operation
        Data data = Accessors.getSerializationService(node).toData("data");
        int partitionId = Accessors.getPartitionService(node).getPartitionId(data);

        MapOperation operation = new SetOperation("test-map", data, data) {
            @Override
            protected void innerBeforeRun() throws Exception {
                super.innerBeforeRun();
                sleepAtLeastSeconds(2);
            }
        };
        operation.setNodeEngine(Accessors.getNodeEngineImpl(node));
        operation.setPartitionId(partitionId);
        operation.setServiceName(MapService.SERVICE_NAME);
        operation.setOperationResponseHandler((op, response) -> {
            expectedResponse.set(response);
            latch.countDown();
        });

        // set op times out after 1 second
        OperationAccessor.setCallTimeout(operation, 1000);
        OperationAccessor.setInvocationTime(operation, Clock.currentTimeMillis());
        Accessors.getOperationService(node).execute(operation);

        // wait operation end
        assertOpenEventually(latch);
        assertInstanceOf(CallTimeoutResponse.class, expectedResponse.get());
    }