A DESCRIPTION OF THE REQUEST :
In case when one worker thread is reused, it may happen that this thread would accumulate thread locals from executed tasks, which could not be used anymore

or, even worse, prevent GC collect objects that they hold. There is should be a way to tear off accumulated thread locals (as a whole) on task completion.

Also It would be good if task would work in clean context and use inheritableThreadLocals from thread that created and scheduled the task, but not from

thread that created workers; i.e. as if it works in newly created thread.

This example demonstrates problem (OOME):
public class Main {
	private static final int HUGE_BLOCK_SIZE = (int) (Runtime.getRuntime().maxMemory() / 2 + 1);

	private static ThreadLocal tls = new ThreadLocal<Object>() {
		protected Object initialValue() {
			System.out.println(Thread.currentThread());
			return new byte[HUGE_BLOCK_SIZE];
		}
	};

	private static ExecutorService executor = Executors.newFixedThreadPool(2);
	static void execute(Runnable task) throws ExecutionException, InterruptedException {
		System.out.println("begin");
		Runnable r = task;
		Future<?> future = executor.submit(r);
		future.get(); // wait for completion
		System.out.println("end");
	}

	public static void main(String[] args) throws Exception {
		Runnable task = new Runnable() {
			public void run() {
				tls.get();
			}
		};
		try {
			// the first time the task is executed in worker #1
			// huge object is created and put it to threadLocals of worker #1
			execute(task);
			// the second time the task is executed in worker #2
			// huge object cannot be created since worker #1 still holds its threadLocals
			execute(task);
		} finally {
			executor.shutdown();
		}
	}

}

Please note that no problem appears if method execute() would be implemented in way that always creates new worker thread. I.e. like this:
	static void execute(Runnable task) throws InterruptedException {
		System.out.println("begin");
		Thread thread = new Thread(task);
		thread.start();
		thread.join();
		System.out.println("end");
	}

I propose to create class ThreadLocalScope, which has to clean up thread locals that are created during task execution. It should be possible to use it in

following way:
	static void execute(Runnable task) throws ExecutionException, InterruptedException {
		System.out.println("begin");
		Runnable r = new ThreadLocalScope(task);
		Future<?> future = executor.submit(r);
		future.get(); // wait for completion
		System.out.println("end");
	}

Here is a sample of ThreadLocalScope implementation:
package java.lang;

/**
 * <p>The <tt>ThreadLocalScope</tt> is to prevent polution of map of thread locals,
 * which could take place when one or more worker threads are reused to
 * execute several tasks. In general, after task completion it restores that thread locals
 * that were before the task were executed.</p>
 *
 * <p>The <tt>ThreadLocalScope</tt> could be used in two cases:
 * <ol>
 * <li>As a decorator of task that are to be executed in separate thread
 * <pre>
 * Runnable task = ...
 * threadPool.execute(new ThreadLocalScope(task));
 * </pre>
 * </li>
 * <li>As an executor of task in current thread
 * <pre>
 * Runnable task = ...
 * ThreadLocalScope.run(task);
 * </pre>
 * </li>
 * </ol>
 * </p>
 *
 * <p>Also the <tt>ThreadLocalScope</tt> can work in two modes:
 * <ol>
 * <li>task is executed as it is in a newly created thread. I.e. inheritable thread locals
 * are inherited from the thread that created instance of <tt>ThreadLocalScope</tt> and non-inheritable thread locals are empty.</li>
 * <li>task is executed as it is in a worker thread but changes
 * in thread locals are to be reverted anyway</li>
 * </ol>
 * </p>
 */
public class ThreadLocalScope implements Runnable {
	private final Runnable _target;
	private final boolean _mode;
	private final ThreadLocal.ThreadLocalMap _heritage;

	/**
	 * Similar to
	 * <pre>
	 * new ThreadLocalScope(true, target);
	 * </pre>
	 * @see ThreadLocalScope#ThreadLocalScope(boolean, Runnable)
	 * @param target decorated task
	 * @throws NullPointerException if target is null
	 */
	public ThreadLocalScope(Runnable target) {
		this(true, target);
	}

	/**
	 * @param inheritThreadLocalsFromCreator mode, if it is true the task is executed as it is in a newly created thread,
	 * otherwise the task is executed as it is in a worker thread.
	 * @param target decorated task
	 * @throws NullPointerException if target is null
	 */
	public ThreadLocalScope(boolean inheritThreadLocalsFromCreator, Runnable target) {
		if (target == null)
			throw new NullPointerException("target");
		_target = target;
		_mode = inheritThreadLocalsFromCreator;
		_heritage = inheritThreadLocalsFromCreator? copy(Thread.currentThread().inheritableThreadLocals): null;
	}

	static ThreadLocal.ThreadLocalMap copy(ThreadLocal.ThreadLocalMap map) {
		return (map == null)? null: ThreadLocal.createInheritedMap(map);
	}

	public void run() {
		run(_heritage, _mode, _target);
	}

	/**
	 * Executes task in current thread.
	 * Similar to
	 * <pre>
	 * ThreadLocalScope.run(false, target);
	 * </pre>
	 * @param target decorated task
	 * @throws NullPointerException if target is null
	 */
	public static void run(Runnable target) {
		run(false, target);
	}

	/**
	 * Executes task in current thread.
	 * @param inheritThreadLocalsFromCreator mode, if it is null the task is executed as it is in a newly created thread
	 * (i.e. non-inheritable thread locals are empty), the task is executed as it is in current thread (i.e. all thread locals
	 * are accessible but any changes are reverted after task completion).
	 * @param target decorated task
	 * @throws NullPointerException if target is null
	 */
	public static void run(boolean inheritThreadLocalsFromCreator, Runnable target) {
		if (target == null)
			throw new NullPointerException("target");
		ThreadLocal.ThreadLocalMap heritage = inheritThreadLocalsFromCreator? Thread.currentThread().inheritableThreadLocals: null;
		run(heritage, inheritThreadLocalsFromCreator, target);
	}

	static void run(ThreadLocal.ThreadLocalMap heritage, boolean inheritThreadLocalsFromCreator, Runnable target) {
		Thread current = Thread.currentThread();
		ThreadLocal.ThreadLocalMap inheritableThreadLocals = current.inheritableThreadLocals;
		ThreadLocal.ThreadLocalMap threadLocals = current.threadLocals;
		if (inheritThreadLocalsFromCreator) {
			current.inheritableThreadLocals = copy(heritage);
			current.threadLocals = null;
		} else {
			ThreadLocal.ThreadLocalMap tmp = copy(inheritableThreadLocals);
			current.threadLocals = copy(threadLocals);
			current.inheritableThreadLocals = tmp;
		}
		try {
			target.run();
		} finally {
			current.inheritableThreadLocals = inheritableThreadLocals;
			current.threadLocals = threadLocals;
		}
	}

}


JUSTIFICATION :
see description

EXPECTED VERSUS ACTUAL BEHAVIOR :
EXPECTED -
begin
Thread[Thread-0,5,main]
end
begin
Thread[Thread-1,5,main]
end

ACTUAL -
begin
Thread[pool-1-thread-1,5,main]
end
begin
Thread[pool-1-thread-2,5,main]
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.OutOfMemoryError: Java heap space
	at java.util.concurrent.FutureTask$Sync.innerGet(Unknown Source)
	at java.util.concurrent.FutureTask.get(Unknown Source)
	at org.shejko.tlscope.Main.execute(Main.java:42)
	at org.shejko.tlscope.Main.main(Main.java:58)
Caused by: java.lang.OutOfMemoryError: Java heap space


---------- BEGIN SOURCE ----------
see description
---------- END SOURCE ----------

CUSTOMER SUBMITTED WORKAROUND :
1. avoid using thread pools for tasks that use thread locals
2. use explicit ThreadLocal.remove() method when task is completed, this is not always practical since some thread locals could be created in internals of third-party libraries and it is not common practice to expose instances of ThreadLocal for clean up.