exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerDecorator.java - drill - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.drill.exec.physical.impl.partitionsender;

 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.LockSupport;

 import org.apache.drill.exec.ops.FragmentContext;
 import org.apache.drill.exec.ops.OperatorStats;
 import org.apache.drill.exec.ops.QueryCancelledException;
 import org.apache.drill.exec.record.RecordBatch;
 import org.apache.drill.exec.testing.ControlsInjector;
 import org.apache.drill.exec.testing.ControlsInjectorFactory;
 import org.apache.drill.exec.testing.CountDownLatchInjection;

 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.MoreExecutors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * Decorator class to hide multiple Partitioner existence from the caller since
  * this class involves multithreaded processing of incoming batches as well as
  * flushing it needs special handling of OperatorStats - stats since stats are
  * not suitable for use in multithreaded environment The algorithm to figure out
  * processing versus wait time is based on following formula: totalWaitTime =
  * totalAllPartitionersProcessingTime - max(sum(processingTime) by partitioner)
  */
 public final class PartitionerDecorator {
   private static final Logger logger = LoggerFactory.getLogger(PartitionerDecorator.class);
   private static final ControlsInjector injector = ControlsInjectorFactory.getInjector(PartitionerDecorator.class);

   private final List<Partitioner> partitioners;
   private final OperatorStats stats;
   private final ExecutorService executor;
   private final FragmentContext context;
   private final Thread thread;
   private final boolean enableParallelTaskExecution;

   PartitionerDecorator(List<Partitioner> partitioners, OperatorStats stats, FragmentContext context) {
     this(partitioners, stats, context, partitioners.size() > 1);
   }

   PartitionerDecorator(List<Partitioner> partitioners, OperatorStats stats, FragmentContext context, boolean enableParallelTaskExecution) {
     this.partitioners = partitioners;
     this.stats = stats;
     this.context = context;
     this.enableParallelTaskExecution = enableParallelTaskExecution;
     executor =  enableParallelTaskExecution ?  context.getExecutor() : MoreExecutors.newDirectExecutorService();
     thread = Thread.currentThread();
   }

   /**
    * partitionBatch - decorator method to call real Partitioner(s) to process incoming batch
    * uses either threading or not threading approach based on number Partitioners
    * @param incoming
    * @throws ExecutionException
    */
   public void partitionBatch(final RecordBatch incoming) throws ExecutionException {
     executeMethodLogic(new PartitionBatchHandlingClass(incoming));
   }

   /**
    * flushOutgoingBatches - decorator to call real Partitioner(s) flushOutgoingBatches
    * @param isLastBatch
    * @param schemaChanged
    * @throws ExecutionException
    */
   public void flushOutgoingBatches(final boolean isLastBatch, final boolean schemaChanged) throws ExecutionException {
     executeMethodLogic(new FlushBatchesHandlingClass(isLastBatch, schemaChanged));
   }

   /**
    * decorator method to call multiple Partitioners initialize()
    */
   public void initialize() {
     for (Partitioner part : partitioners ) {
       part.initialize();
     }
   }

   /**
    * decorator method to call multiple Partitioners clear()
    */
   public void clear() {
     for (Partitioner part : partitioners ) {
       part.clear();
     }
   }

   /**
    * Helper method to get PartitionOutgoingBatch based on the index
    * since we may have more then one Partitioner
    * As number of Partitioners should be very small AND this method it used very rarely,
    * so it is OK to loop in order to find right partitioner
    * @param index - index of PartitionOutgoingBatch
    * @return PartitionOutgoingBatch
    */
   public PartitionOutgoingBatch getOutgoingBatches(int index) {
     for (Partitioner part : partitioners ) {
       PartitionOutgoingBatch outBatch = part.getOutgoingBatch(index);
       if ( outBatch != null ) {
         return outBatch;
       }
     }
     return null;
   }

   List<Partitioner> getPartitioners() {
     return partitioners;
   }

   /**
    * Helper to execute the different methods wrapped into same logic
    * @param iface
    * @throws ExecutionException
    */
   @VisibleForTesting
   void executeMethodLogic(final GeneralExecuteIface iface) throws ExecutionException {
     // To simulate interruption of main fragment thread and interrupting the partition threads, create a
     // CountDownInject latch. Partitioner threads await on the latch and main fragment thread counts down or
     // interrupts waiting threads. This makes sure that we are actually interrupting the blocked partitioner threads.
     try (CountDownLatchInjection testCountDownLatch = injector.getLatch(context.getExecutionControls(), "partitioner-sender-latch")) {
       testCountDownLatch.initialize(1);
       final AtomicInteger count = new AtomicInteger();
       List<PartitionerTask> partitionerTasks = new ArrayList<>(partitioners.size());
       ExecutionException executionException = null;
       // start waiting on main stats to adjust by sum(max(processing)) at the end
       startWait();
       try {
         partitioners.forEach(partitioner -> createAndExecute(iface, testCountDownLatch, count, partitionerTasks, partitioner));
         // Wait for main fragment interruption.
         injector.injectInterruptiblePause(context.getExecutionControls(), "wait-for-fragment-interrupt", logger);
         testCountDownLatch.countDown();
       } catch (InterruptedException e) {
         logger.warn("fragment thread interrupted", e);
         throw new QueryCancelledException();
       } catch (RejectedExecutionException e) {
         logger.warn("Failed to execute partitioner tasks. Execution service down?", e);
         executionException = new ExecutionException(e);
       } finally {
         await(count, partitionerTasks);
         stopWait();
         processPartitionerTasks(partitionerTasks, executionException);
       }
     }
   }

   private void createAndExecute(GeneralExecuteIface iface, CountDownLatchInjection testCountDownLatch, AtomicInteger count,
       List<PartitionerTask> partitionerTasks, Partitioner partitioner) {
     PartitionerTask partitionerTask = new PartitionerTask(this, iface, partitioner, count, testCountDownLatch);
     executor.execute(partitionerTask);
     partitionerTasks.add(partitionerTask);
     count.incrementAndGet();
   }

   /**
    * Wait for completion of all partitioner tasks.
    * @param count current number of task not yet completed
    * @param partitionerTasks list of partitioner tasks submitted for execution
    */
   private void await(AtomicInteger count, List<PartitionerTask> partitionerTasks) {
     boolean cancelled = false;
     while (count.get() > 0) {
       if (context.getExecutorState().shouldContinue() || cancelled) {
         LockSupport.park();
       } else {
         logger.warn("Cancelling fragment {} partitioner tasks...", context.getFragIdString());
         partitionerTasks.forEach(partitionerTask -> partitionerTask.cancel(true));
         cancelled = true;
       }
     }
   }

   private void startWait() {
     if (enableParallelTaskExecution) {
       stats.startWait();
     }
   }

   private void stopWait() {
     if (enableParallelTaskExecution) {
       stats.stopWait();
     }
   }

   private void processPartitionerTasks(List<PartitionerTask> partitionerTasks, ExecutionException executionException) throws ExecutionException {
     long maxProcessTime = 0l;
     for (PartitionerTask partitionerTask : partitionerTasks) {
       ExecutionException e = partitionerTask.getException();
       if (e != null) {
         if (executionException == null) {
           executionException = e;
         } else {
           executionException.getCause().addSuppressed(e.getCause());
         }
       }
       if (executionException == null) {
         final OperatorStats localStats = partitionerTask.getStats();
         // find out max Partitioner processing time
         if (enableParallelTaskExecution) {
           long currentProcessingNanos = localStats.getProcessingNanos();
           maxProcessTime = (currentProcessingNanos > maxProcessTime) ? currentProcessingNanos : maxProcessTime;
         } else {
           maxProcessTime += localStats.getWaitNanos();
         }
         stats.mergeMetrics(localStats);
       }
     }
     if (executionException != null) {
       throw executionException;
     }
     // scale down main stats wait time based on calculated processing time
     // since we did not wait for whole duration of above execution
     if (enableParallelTaskExecution) {
       stats.adjustWaitNanos(-maxProcessTime);
     } else {
       stats.adjustWaitNanos(maxProcessTime);
     }
   }

   /**
    * Helper interface to generalize functionality executed in the thread
    * since it is absolutely the same for partitionBatch and flushOutgoingBatches
    * protected is for testing purposes
    */
   protected interface GeneralExecuteIface {
     void execute(Partitioner partitioner) throws IOException;
   }

   /**
    * Class to handle running partitionBatch method
    *
    */
   private static class PartitionBatchHandlingClass implements GeneralExecuteIface {

     private final RecordBatch incoming;

     PartitionBatchHandlingClass(RecordBatch incoming) {
       this.incoming = incoming;
     }

     @Override
     public void execute(Partitioner part) throws IOException {
       part.partitionBatch(incoming);
     }
   }

   /**
    * Class to handle running flushOutgoingBatches method
    *
    */
   private static class FlushBatchesHandlingClass implements GeneralExecuteIface {

     private final boolean isLastBatch;
     private final boolean schemaChanged;

     public FlushBatchesHandlingClass(boolean isLastBatch, boolean schemaChanged) {
       this.isLastBatch = isLastBatch;
       this.schemaChanged = schemaChanged;
     }

     @Override
     public void execute(Partitioner part) throws IOException {
       part.flushOutgoingBatches(isLastBatch, schemaChanged);
     }
   }

   /**
    * Helper class to wrap Runnable with cancellation and waiting for completion support
    *
    */
   private static class PartitionerTask implements Runnable {

     private enum STATE {
       NEW,
       COMPLETING,
       NORMAL,
       EXCEPTIONAL,
       CANCELLED,
       INTERRUPTING,
       INTERRUPTED
     }

     private final AtomicReference<STATE> state;
     private final AtomicReference<Thread> runner;
     private final PartitionerDecorator partitionerDecorator;
     private final AtomicInteger count;

     private final GeneralExecuteIface iface;
     private final Partitioner partitioner;
     private final CountDownLatchInjection testCountDownLatch;

     private volatile ExecutionException exception;

     public PartitionerTask(PartitionerDecorator partitionerDecorator, GeneralExecuteIface iface, Partitioner partitioner, AtomicInteger count, CountDownLatchInjection testCountDownLatch) {
       state = new AtomicReference<>(STATE.NEW);
       runner = new AtomicReference<>();
       this.partitionerDecorator = partitionerDecorator;
       this.iface = iface;
       this.partitioner = partitioner;
       this.count = count;
       this.testCountDownLatch = testCountDownLatch;
     }

     @Override
     public void run() {
       final Thread thread = Thread.currentThread();
       if (runner.compareAndSet(null, thread)) {
         final String name = thread.getName();
         thread.setName(String.format("Partitioner-%s-%d", partitionerDecorator.thread.getName(), thread.getId()));
         final OperatorStats localStats = partitioner.getStats();
         localStats.clear();
         localStats.startProcessing();
         ExecutionException executionException = null;
         try {
           // Test only - Pause until interrupted by fragment thread
           testCountDownLatch.await();
           if (state.get() == STATE.NEW) {
             iface.execute(partitioner);
           }
         } catch (InterruptedException e) {
           if (state.compareAndSet(STATE.NEW, STATE.INTERRUPTED)) {
             logger.warn("Partitioner Task interrupted during the run", e);
           }
         } catch (Throwable t) {
           executionException = new ExecutionException(t);
         } finally {
           if (state.compareAndSet(STATE.NEW, STATE.COMPLETING)) {
             if (executionException == null) {
               localStats.stopProcessing();
               state.lazySet(STATE.NORMAL);
             } else {
               exception = executionException;
               state.lazySet(STATE.EXCEPTIONAL);
             }
           }
           if (count.decrementAndGet() == 0) {
             LockSupport.unpark(partitionerDecorator.thread);
           }
           thread.setName(name);
           while (state.get() == STATE.INTERRUPTING) {
             Thread.yield();
           }
           // Clear interrupt flag
           Thread.interrupted();
         }
       }
     }

     void cancel(boolean mayInterruptIfRunning) {
       Preconditions.checkState(Thread.currentThread() == partitionerDecorator.thread,
           String.format("PartitionerTask can be cancelled only from the main %s thread", partitionerDecorator.thread.getName()));
       if (runner.compareAndSet(null, partitionerDecorator.thread)) {
         if (partitionerDecorator.executor instanceof ThreadPoolExecutor) {
           ((ThreadPoolExecutor)partitionerDecorator.executor).remove(this);
         }
         count.decrementAndGet();
       } else {
         if (mayInterruptIfRunning) {
           if (state.compareAndSet(STATE.NEW, STATE.INTERRUPTING)) {
             try {
               runner.get().interrupt();
             } finally {
               state.lazySet(STATE.INTERRUPTED);
             }
           }
         } else {
           state.compareAndSet(STATE.NEW, STATE.CANCELLED);
         }
       }
     }

     public ExecutionException getException() {
       return exception;
     }

     public OperatorStats getStats() {
       return partitioner.getStats();
     }
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.drill.exec.physical.impl.partitionsender;

	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.List;
	import java.util.concurrent.ExecutionException;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.RejectedExecutionException;
	import java.util.concurrent.ThreadPoolExecutor;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.atomic.AtomicReference;
	import java.util.concurrent.locks.LockSupport;

	import org.apache.drill.exec.ops.FragmentContext;
	import org.apache.drill.exec.ops.OperatorStats;
	import org.apache.drill.exec.ops.QueryCancelledException;
	import org.apache.drill.exec.record.RecordBatch;
	import org.apache.drill.exec.testing.ControlsInjector;
	import org.apache.drill.exec.testing.ControlsInjectorFactory;
	import org.apache.drill.exec.testing.CountDownLatchInjection;

	import com.google.common.annotations.VisibleForTesting;
	import com.google.common.base.Preconditions;
	import com.google.common.util.concurrent.MoreExecutors;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* Decorator class to hide multiple Partitioner existence from the caller since
	* this class involves multithreaded processing of incoming batches as well as
	* flushing it needs special handling of OperatorStats - stats since stats are
	* not suitable for use in multithreaded environment The algorithm to figure out
	* processing versus wait time is based on following formula: totalWaitTime =
	* totalAllPartitionersProcessingTime - max(sum(processingTime) by partitioner)
	*/
	public final class PartitionerDecorator {
	private static final Logger logger = LoggerFactory.getLogger(PartitionerDecorator.class);
	private static final ControlsInjector injector = ControlsInjectorFactory.getInjector(PartitionerDecorator.class);

	private final List<Partitioner> partitioners;
	private final OperatorStats stats;
	private final ExecutorService executor;
	private final FragmentContext context;
	private final Thread thread;
	private final boolean enableParallelTaskExecution;

	PartitionerDecorator(List<Partitioner> partitioners, OperatorStats stats, FragmentContext context) {
	this(partitioners, stats, context, partitioners.size() > 1);
	}

	PartitionerDecorator(List<Partitioner> partitioners, OperatorStats stats, FragmentContext context, boolean enableParallelTaskExecution) {
	this.partitioners = partitioners;
	this.stats = stats;
	this.context = context;
	this.enableParallelTaskExecution = enableParallelTaskExecution;
	executor = enableParallelTaskExecution ? context.getExecutor() : MoreExecutors.newDirectExecutorService();
	thread = Thread.currentThread();
	}

	/**
	* partitionBatch - decorator method to call real Partitioner(s) to process incoming batch
	* uses either threading or not threading approach based on number Partitioners
	* @param incoming
	* @throws ExecutionException
	*/
	public void partitionBatch(final RecordBatch incoming) throws ExecutionException {
	executeMethodLogic(new PartitionBatchHandlingClass(incoming));
	}

	/**
	* flushOutgoingBatches - decorator to call real Partitioner(s) flushOutgoingBatches
	* @param isLastBatch
	* @param schemaChanged
	* @throws ExecutionException
	*/
	public void flushOutgoingBatches(final boolean isLastBatch, final boolean schemaChanged) throws ExecutionException {
	executeMethodLogic(new FlushBatchesHandlingClass(isLastBatch, schemaChanged));
	}

	/**
	* decorator method to call multiple Partitioners initialize()
	*/
	public void initialize() {
	for (Partitioner part : partitioners ) {
	part.initialize();
	}
	}

	/**
	* decorator method to call multiple Partitioners clear()
	*/
	public void clear() {
	for (Partitioner part : partitioners ) {
	part.clear();
	}
	}

	/**
	* Helper method to get PartitionOutgoingBatch based on the index
	* since we may have more then one Partitioner
	* As number of Partitioners should be very small AND this method it used very rarely,
	* so it is OK to loop in order to find right partitioner
	* @param index - index of PartitionOutgoingBatch
	* @return PartitionOutgoingBatch
	*/
	public PartitionOutgoingBatch getOutgoingBatches(int index) {
	for (Partitioner part : partitioners ) {
	PartitionOutgoingBatch outBatch = part.getOutgoingBatch(index);
	if ( outBatch != null ) {
	return outBatch;
	}
	}
	return null;
	}

	List<Partitioner> getPartitioners() {
	return partitioners;
	}

	/**
	* Helper to execute the different methods wrapped into same logic
	* @param iface
	* @throws ExecutionException
	*/
	@VisibleForTesting
	void executeMethodLogic(final GeneralExecuteIface iface) throws ExecutionException {
	// To simulate interruption of main fragment thread and interrupting the partition threads, create a
	// CountDownInject latch. Partitioner threads await on the latch and main fragment thread counts down or
	// interrupts waiting threads. This makes sure that we are actually interrupting the blocked partitioner threads.
	try (CountDownLatchInjection testCountDownLatch = injector.getLatch(context.getExecutionControls(), "partitioner-sender-latch")) {
	testCountDownLatch.initialize(1);
	final AtomicInteger count = new AtomicInteger();
	List<PartitionerTask> partitionerTasks = new ArrayList<>(partitioners.size());
	ExecutionException executionException = null;
	// start waiting on main stats to adjust by sum(max(processing)) at the end
	startWait();
	try {
	partitioners.forEach(partitioner -> createAndExecute(iface, testCountDownLatch, count, partitionerTasks, partitioner));
	// Wait for main fragment interruption.
	injector.injectInterruptiblePause(context.getExecutionControls(), "wait-for-fragment-interrupt", logger);
	testCountDownLatch.countDown();
	} catch (InterruptedException e) {
	logger.warn("fragment thread interrupted", e);
	throw new QueryCancelledException();
	} catch (RejectedExecutionException e) {
	logger.warn("Failed to execute partitioner tasks. Execution service down?", e);
	executionException = new ExecutionException(e);
	} finally {
	await(count, partitionerTasks);
	stopWait();
	processPartitionerTasks(partitionerTasks, executionException);
	}
	}
	}

	private void createAndExecute(GeneralExecuteIface iface, CountDownLatchInjection testCountDownLatch, AtomicInteger count,
	List<PartitionerTask> partitionerTasks, Partitioner partitioner) {
	PartitionerTask partitionerTask = new PartitionerTask(this, iface, partitioner, count, testCountDownLatch);
	executor.execute(partitionerTask);
	partitionerTasks.add(partitionerTask);
	count.incrementAndGet();
	}

	/**
	* Wait for completion of all partitioner tasks.
	* @param count current number of task not yet completed
	* @param partitionerTasks list of partitioner tasks submitted for execution
	*/
	private void await(AtomicInteger count, List<PartitionerTask> partitionerTasks) {
	boolean cancelled = false;
	while (count.get() > 0) {
	if (context.getExecutorState().shouldContinue() \|\| cancelled) {
	LockSupport.park();
	} else {
	logger.warn("Cancelling fragment {} partitioner tasks...", context.getFragIdString());
	partitionerTasks.forEach(partitionerTask -> partitionerTask.cancel(true));
	cancelled = true;
	}
	}
	}

	private void startWait() {
	if (enableParallelTaskExecution) {
	stats.startWait();
	}
	}

	private void stopWait() {
	if (enableParallelTaskExecution) {
	stats.stopWait();
	}
	}

	private void processPartitionerTasks(List<PartitionerTask> partitionerTasks, ExecutionException executionException) throws ExecutionException {
	long maxProcessTime = 0l;
	for (PartitionerTask partitionerTask : partitionerTasks) {
	ExecutionException e = partitionerTask.getException();
	if (e != null) {
	if (executionException == null) {
	executionException = e;
	} else {
	executionException.getCause().addSuppressed(e.getCause());
	}
	}
	if (executionException == null) {
	final OperatorStats localStats = partitionerTask.getStats();
	// find out max Partitioner processing time
	if (enableParallelTaskExecution) {
	long currentProcessingNanos = localStats.getProcessingNanos();
	maxProcessTime = (currentProcessingNanos > maxProcessTime) ? currentProcessingNanos : maxProcessTime;
	} else {
	maxProcessTime += localStats.getWaitNanos();
	}
	stats.mergeMetrics(localStats);
	}
	}
	if (executionException != null) {
	throw executionException;
	}
	// scale down main stats wait time based on calculated processing time
	// since we did not wait for whole duration of above execution
	if (enableParallelTaskExecution) {
	stats.adjustWaitNanos(-maxProcessTime);
	} else {
	stats.adjustWaitNanos(maxProcessTime);
	}
	}

	/**
	* Helper interface to generalize functionality executed in the thread
	* since it is absolutely the same for partitionBatch and flushOutgoingBatches
	* protected is for testing purposes
	*/
	protected interface GeneralExecuteIface {
	void execute(Partitioner partitioner) throws IOException;
	}

	/**
	* Class to handle running partitionBatch method
	*
	*/
	private static class PartitionBatchHandlingClass implements GeneralExecuteIface {

	private final RecordBatch incoming;

	PartitionBatchHandlingClass(RecordBatch incoming) {
	this.incoming = incoming;
	}

	@Override
	public void execute(Partitioner part) throws IOException {
	part.partitionBatch(incoming);
	}
	}

	/**
	* Class to handle running flushOutgoingBatches method
	*
	*/
	private static class FlushBatchesHandlingClass implements GeneralExecuteIface {

	private final boolean isLastBatch;
	private final boolean schemaChanged;

	public FlushBatchesHandlingClass(boolean isLastBatch, boolean schemaChanged) {
	this.isLastBatch = isLastBatch;
	this.schemaChanged = schemaChanged;
	}

	@Override
	public void execute(Partitioner part) throws IOException {
	part.flushOutgoingBatches(isLastBatch, schemaChanged);
	}
	}

	/**
	* Helper class to wrap Runnable with cancellation and waiting for completion support
	*
	*/
	private static class PartitionerTask implements Runnable {

	private enum STATE {
	NEW,
	COMPLETING,
	NORMAL,
	EXCEPTIONAL,
	CANCELLED,
	INTERRUPTING,
	INTERRUPTED
	}

	private final AtomicReference<STATE> state;
	private final AtomicReference<Thread> runner;
	private final PartitionerDecorator partitionerDecorator;
	private final AtomicInteger count;

	private final GeneralExecuteIface iface;
	private final Partitioner partitioner;
	private final CountDownLatchInjection testCountDownLatch;

	private volatile ExecutionException exception;

	public PartitionerTask(PartitionerDecorator partitionerDecorator, GeneralExecuteIface iface, Partitioner partitioner, AtomicInteger count, CountDownLatchInjection testCountDownLatch) {
	state = new AtomicReference<>(STATE.NEW);
	runner = new AtomicReference<>();
	this.partitionerDecorator = partitionerDecorator;
	this.iface = iface;
	this.partitioner = partitioner;
	this.count = count;
	this.testCountDownLatch = testCountDownLatch;
	}

	@Override
	public void run() {
	final Thread thread = Thread.currentThread();
	if (runner.compareAndSet(null, thread)) {
	final String name = thread.getName();
	thread.setName(String.format("Partitioner-%s-%d", partitionerDecorator.thread.getName(), thread.getId()));
	final OperatorStats localStats = partitioner.getStats();
	localStats.clear();
	localStats.startProcessing();
	ExecutionException executionException = null;
	try {
	// Test only - Pause until interrupted by fragment thread
	testCountDownLatch.await();
	if (state.get() == STATE.NEW) {
	iface.execute(partitioner);
	}
	} catch (InterruptedException e) {
	if (state.compareAndSet(STATE.NEW, STATE.INTERRUPTED)) {
	logger.warn("Partitioner Task interrupted during the run", e);
	}
	} catch (Throwable t) {
	executionException = new ExecutionException(t);
	} finally {
	if (state.compareAndSet(STATE.NEW, STATE.COMPLETING)) {
	if (executionException == null) {
	localStats.stopProcessing();
	state.lazySet(STATE.NORMAL);
	} else {
	exception = executionException;
	state.lazySet(STATE.EXCEPTIONAL);
	}
	}
	if (count.decrementAndGet() == 0) {
	LockSupport.unpark(partitionerDecorator.thread);
	}
	thread.setName(name);
	while (state.get() == STATE.INTERRUPTING) {
	Thread.yield();
	}
	// Clear interrupt flag
	Thread.interrupted();
	}
	}
	}

	void cancel(boolean mayInterruptIfRunning) {
	Preconditions.checkState(Thread.currentThread() == partitionerDecorator.thread,
	String.format("PartitionerTask can be cancelled only from the main %s thread", partitionerDecorator.thread.getName()));
	if (runner.compareAndSet(null, partitionerDecorator.thread)) {
	if (partitionerDecorator.executor instanceof ThreadPoolExecutor) {
	((ThreadPoolExecutor)partitionerDecorator.executor).remove(this);
	}
	count.decrementAndGet();
	} else {
	if (mayInterruptIfRunning) {
	if (state.compareAndSet(STATE.NEW, STATE.INTERRUPTING)) {
	try {
	runner.get().interrupt();
	} finally {
	state.lazySet(STATE.INTERRUPTED);
	}
	}
	} else {
	state.compareAndSet(STATE.NEW, STATE.CANCELLED);
	}
	}
	}

	public ExecutionException getException() {
	return exception;
	}

	public OperatorStats getStats() {
	return partitioner.getStats();
	}
	}
	}