package org.embulk.output.sftp.utils;

import com.google.common.base.Optional;
import org.embulk.spi.Exec;
import org.slf4j.Logger;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public abstract class TimedCallable<V> implements Callable<V>
{
    private static final ExecutorService THREAD_POOL = Executors.newCachedThreadPool();
    private Logger logger = Exec.getLogger(getClass());

    public V callNonInterruptible(long timeout, TimeUnit timeUnit)
    {
        try {
            return call(timeout, timeUnit);
        }
        catch (Exception e) {
            logger.warn("Failed with exception {}: {}", e.getClass(), Optional.fromNullable(e.getMessage()).or(""));
            return null;
        }
    }

    public V call(long timeout, TimeUnit timeUnit)
            throws InterruptedException, ExecutionException, TimeoutException
    {
        FutureTask<V> task = new FutureTask<>(this);
        try {
            THREAD_POOL.execute(task);
            return task.get(timeout, timeUnit);
        }
        catch (TimeoutException e) {
            logger.warn("Execution timed out, re-throwing ...");
            task.cancel(true);
            throw e;
        }
    }
}

@@ -36,48 +36,8 @@
 s;%0A%0A
-import com.google.common.base.Optional;%0A
 impo
@@ -790,97 +790,51 @@
 rn(%22
-Failed with exception %7B%7D: %7B%7D%22, e.getClass(), Optional.fromNullable(e.getMessage()).or(%22%22)
+Time-out call failed, ignore and resume%22, e
 );%0A 
@@ -1185,152 +1185,47 @@
     
-catch (TimeoutException e) %7B%0A            logger.warn(%22Execution timed out, re-throwing ...%22);%0A            task.cancel(true);%0A            throw e
+finally %7B%0A            task.cancel(true)
 ;%0A  

Fix a bug of `task.cancel()`, which keeps the task running forever when no exception is thrown