We are writing an app that uses a custom extension of the ThreadPoolExecutor to process a bunch of Runnables. In developing the app, we have run into OutOfMemoryErrors that occurred in our Runnable classes. But, instead of calling our afterExecute() like we would expect, the ThreadPoolExecutor just kept on trucking along.
I have pared down the code to a small, repeatable app.
ThreadPoolTest.java:
package org.codechimp.threadpool;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolTest {
/**
* @param args
*/
public static void main(String[] args) {
LinkedBlockingQueue<Runnable> threadQueue = new LinkedBlockingQueue<Runnable>();
ThreadPoolExecutor executor = new MyThreadPoolExecutor(10, Integer.MAX_VALUE, 2000, TimeUnit.MILLISECONDS, threadQueue);
// Create a bunch of Runnables now...
for (int i = 0; i < 10000; i++) {
executor.execute(new MyRunnable(i));
if (i % 100 == 0) {
System.out.println("Queued " + i + " Runnables");
}
if (i % 1000 == 0) {
try {
Thread.sleep(1);
} catch (InterruptedException ignored) { }
}
}
System.out.println("Done queing the Runnables.");
while (!executor.isTerminated() && !executor.isTerminating() && executor.getQueue().size() > 0) {
System.out.println(executor.getQueue().size() + " Runnables in the queue.");
try {
Thread.sleep(1000);
} catch (InterruptedException ignored) { }
}
System.out.println("Runnable queue has stopped processing.");
executor.shutdown();
try {
executor.awaitTermination(10, TimeUnit.MINUTES);
} catch (InterruptedException ignored) { }
System.out.println("Shutdown completed...exiting");
}
}
MyThreadPoolExecutor.java:
package org.codechimp.threadpool;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class MyThreadPoolExecutor extends ThreadPoolExecutor {
public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
}
public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, handler);
}
public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory);
}
public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
if (t != null) {
System.out.println("We got an error: " + t);
int remaining = this.shutdownNow().size();
System.out.println(remaining + " Runnables left in the queue");
}
}
}
MyRunnable.java
package org.codechimp.threadpool;
import org.apache.commons.lang.math.RandomUtils;
public class MyRunnable implements Runnable {
private int runnableNumber;
private int counter = 0;
public MyRunnable(int number) {
this.runnableNumber = number;
}
/**
* Simple runnable that generates an OutOfMemoryError after the 1000th loop
*/
public void run() {
while (counter < 1000) {
counter++;
if (counter % 100 == 0) {
System.out.println("\tRunnable " + this.runnableNumber + " reached " + this.counter + ".");
}
if (this.runnableNumber == 15 && this.counter % 200 == 0) {
throw new OutOfMemoryError("This is a test!");
}
int wait = RandomUtils.nextInt(100);
if (wait > 0) {
try {
//System.out.println("\tRunnable " + this.runnableNumber + " waiting " + wait + ".");
Thread.sleep(wait);
} catch (InterruptedException e) {
throw new RuntimeException("Thread.sleep() failed", e);
}
}
}
}
}
This is a super-simple example that will create 10k of MyRunnables into the MyThreadPoolExectuor, printing out some status messages as the counters increase. The 16th runnable (number 15, counts from 0) will throw an OutOfMemoryError on the 200th increment. MyThreadPoolExecutor's afterExecute() will print a message if it gets a Throwable. When I run it under Java 6 and Java 7, it never prints this message.
How do we make the app bail on ALL Throwables? We really want to just terminate things at that point.
Update/Edit:
I am updating this as there seems to be some confusion as to what I am asking. I know the error gets printed out. The problem isn't that ThreadPoolExecutor isn't printing OutOfMemoryError, the problem is, as stated in the title and in the question I ask in the end, that the afterExecute() is not being called for Errors. Since OutOfMemoryError is a sub-class of Error, that means when one occurs I have NO WAY to stop the code.
Again, please read what the code is trying to do. It is most certainly trying to "handle" the error. It is in fact trying to stop the ThreadPoolExecutor by calling shutdownNow() on it. But, since the Error being generated is being suppressed somehow, that bit of code is not being executed. As a result, the application simply keeps chugging along, ignoring the fact that it's spewing OOME all over the place.
Again, the question:
How do I detect a Runnable has recieved an Error (OutOfMemoryError or other) and terminate the ThreadPoolExecutor so the app will simply stop in it's tracks?
