You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2018/03/08 16:51:01 UTC
[2/5] flink git commit: [hotfix] [taskmanager] Fix checkstyle in Task
and TaskTest
[hotfix] [taskmanager] Fix checkstyle in Task and TaskTest
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/53168d00
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/53168d00
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/53168d00
Branch: refs/heads/master
Commit: 53168d00cf07f1b93d048f3fc849ce354fa8c7d4
Parents: 236ceb5
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Mar 6 15:54:13 2018 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Mar 8 16:18:04 2018 +0100
----------------------------------------------------------------------
.../apache/flink/runtime/taskmanager/Task.java | 80 ++++++------
.../flink/runtime/taskmanager/TaskTest.java | 126 ++++++++++---------
2 files changed, 106 insertions(+), 100 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/53168d00/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
index 1ecb47a..00fbdfe 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -127,10 +127,10 @@ public class Task implements Runnable, TaskActions, CheckpointListener {
/** The class logger. */
private static final Logger LOG = LoggerFactory.getLogger(Task.class);
- /** The tread group that contains all task threads */
+ /** The tread group that contains all task threads. */
private static final ThreadGroup TASK_THREADS_GROUP = new ThreadGroup("Flink Task Threads");
- /** For atomic state updates */
+ /** For atomic state updates. */
private static final AtomicReferenceFieldUpdater<Task, ExecutionState> STATE_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(Task.class, ExecutionState.class, "executionState");
@@ -138,52 +138,52 @@ public class Task implements Runnable, TaskActions, CheckpointListener {
// Constant fields that are part of the initial Task construction
// ------------------------------------------------------------------------
- /** The job that the task belongs to */
+ /** The job that the task belongs to. */
private final JobID jobId;
- /** The vertex in the JobGraph whose code the task executes */
+ /** The vertex in the JobGraph whose code the task executes. */
private final JobVertexID vertexId;
- /** The execution attempt of the parallel subtask */
+ /** The execution attempt of the parallel subtask. */
private final ExecutionAttemptID executionId;
- /** ID which identifies the slot in which the task is supposed to run */
+ /** ID which identifies the slot in which the task is supposed to run. */
private final AllocationID allocationId;
- /** TaskInfo object for this task */
+ /** TaskInfo object for this task. */
private final TaskInfo taskInfo;
- /** The name of the task, including subtask indexes */
+ /** The name of the task, including subtask indexes. */
private final String taskNameWithSubtask;
- /** The job-wide configuration object */
+ /** The job-wide configuration object. */
private final Configuration jobConfiguration;
- /** The task-specific configuration */
+ /** The task-specific configuration. */
private final Configuration taskConfiguration;
- /** The jar files used by this task */
+ /** The jar files used by this task. */
private final Collection<PermanentBlobKey> requiredJarFiles;
- /** The classpaths used by this task */
+ /** The classpaths used by this task. */
private final Collection<URL> requiredClasspaths;
- /** The name of the class that holds the invokable code */
+ /** The name of the class that holds the invokable code. */
private final String nameOfInvokableClass;
- /** Access to task manager configuration and host names*/
+ /** Access to task manager configuration and host names. */
private final TaskManagerRuntimeInfo taskManagerConfig;
- /** The memory manager to be used by this task */
+ /** The memory manager to be used by this task. */
private final MemoryManager memoryManager;
- /** The I/O manager to be used by this task */
+ /** The I/O manager to be used by this task. */
private final IOManager ioManager;
- /** The BroadcastVariableManager to be used by this task */
+ /** The BroadcastVariableManager to be used by this task. */
private final BroadcastVariableManager broadcastVariableManager;
- /** The manager for state of operators running in this task/slot */
+ /** The manager for state of operators running in this task/slot. */
private final TaskStateManager taskStateManager;
/** Serialized version of the job specific execution configuration (see {@link ExecutionConfig}). */
@@ -195,43 +195,43 @@ public class Task implements Runnable, TaskActions, CheckpointListener {
private final Map<IntermediateDataSetID, SingleInputGate> inputGatesById;
- /** Connection to the task manager */
+ /** Connection to the task manager. */
private final TaskManagerActions taskManagerActions;
- /** Input split provider for the task */
+ /** Input split provider for the task. */
private final InputSplitProvider inputSplitProvider;
- /** Checkpoint notifier used to communicate with the CheckpointCoordinator */
+ /** Checkpoint notifier used to communicate with the CheckpointCoordinator. */
private final CheckpointResponder checkpointResponder;
- /** All listener that want to be notified about changes in the task's execution state */
+ /** All listener that want to be notified about changes in the task's execution state. */
private final List<TaskExecutionStateListener> taskExecutionStateListeners;
- /** The BLOB cache, from which the task can request BLOB files */
+ /** The BLOB cache, from which the task can request BLOB files. */
private final BlobCacheService blobService;
- /** The library cache, from which the task can request its class loader */
+ /** The library cache, from which the task can request its class loader. */
private final LibraryCacheManager libraryCache;
- /** The cache for user-defined files that the invokable requires */
+ /** The cache for user-defined files that the invokable requires. */
private final FileCache fileCache;
- /** The gateway to the network stack, which handles inputs and produced results */
+ /** The gateway to the network stack, which handles inputs and produced results. */
private final NetworkEnvironment network;
- /** The registry of this task which enables live reporting of accumulators */
+ /** The registry of this task which enables live reporting of accumulators. */
private final AccumulatorRegistry accumulatorRegistry;
- /** The thread that executes the task */
+ /** The thread that executes the task. */
private final Thread executingThread;
- /** Parent group for all metrics of this task */
+ /** Parent group for all metrics of this task. */
private final TaskMetricGroup metrics;
- /** Partition producer state checker to request partition states from */
+ /** Partition producer state checker to request partition states from. */
private final PartitionProducerStateChecker partitionProducerStateChecker;
- /** Executor to run future callbacks */
+ /** Executor to run future callbacks. */
private final Executor executor;
// ------------------------------------------------------------------------
@@ -240,19 +240,19 @@ public class Task implements Runnable, TaskActions, CheckpointListener {
// proper happens-before semantics on parallel modification
// ------------------------------------------------------------------------
- /** atomic flag that makes sure the invokable is canceled exactly once upon error */
+ /** atomic flag that makes sure the invokable is canceled exactly once upon error. */
private final AtomicBoolean invokableHasBeenCanceled;
- /** The invokable of this task, if initialized */
+ /** The invokable of this task, if initialized. */
private volatile AbstractInvokable invokable;
- /** The current execution state of the task */
+ /** The current execution state of the task. */
private volatile ExecutionState executionState = ExecutionState.CREATED;
- /** The observed exception, in case the task execution failed */
+ /** The observed exception, in case the task execution failed. */
private volatile Throwable failureCause;
- /** Serial executor for asynchronous calls (checkpoints, etc), lazily initialized */
+ /** Serial executor for asynchronous calls (checkpoints, etc), lazily initialized. */
private volatile ExecutorService asyncCallDispatcher;
/** Initialized from the Flink configuration. May also be set at the ExecutionConfig */
@@ -512,7 +512,7 @@ public class Task implements Runnable, TaskActions, CheckpointListener {
}
/**
- * The core work method that bootstraps the task and executes its code
+ * The core work method that bootstraps the task and executes its code.
*/
@Override
public void run() {
@@ -556,7 +556,7 @@ public class Task implements Runnable, TaskActions, CheckpointListener {
// all resource acquisitions and registrations from here on
// need to be undone in the end
- Map<String, Future<Path>> distributedCacheEntries = new HashMap<String, Future<Path>>();
+ Map<String, Future<Path>> distributedCacheEntries = new HashMap<>();
AbstractInvokable invokable = null;
try {
@@ -743,8 +743,8 @@ public class Task implements Runnable, TaskActions, CheckpointListener {
try {
// check if the exception is unrecoverable
if (ExceptionUtils.isJvmFatalError(t) ||
- (t instanceof OutOfMemoryError && taskManagerConfig.shouldExitJvmOnOutOfMemoryError()))
- {
+ (t instanceof OutOfMemoryError && taskManagerConfig.shouldExitJvmOnOutOfMemoryError())) {
+
// terminate the JVM immediately
// don't attempt a clean shutdown, because we cannot expect the clean shutdown to complete
try {
http://git-wip-us.apache.org/repos/asf/flink/blob/53168d00/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
index 5b33d19..1829e97 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
@@ -29,8 +29,6 @@ import org.apache.flink.runtime.blob.TransientBlobCache;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.concurrent.Executors;
-import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
-import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor;
import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.execution.ExecutionState;
@@ -103,12 +101,12 @@ import static org.mockito.Mockito.when;
* Tests for the Task, which make sure that correct state transitions happen,
* and failures are correctly handled.
*
- * All tests here have a set of mock actors for TaskManager, JobManager, and
+ * <p>All tests here have a set of mock actors for TaskManager, JobManager, and
* execution listener, which simply put the messages in a queue to be picked
* up by the test and validated.
*/
public class TaskTest extends TestLogger {
-
+
private static OneShotLatch awaitLatch;
private static OneShotLatch triggerLatch;
private static OneShotLatch cancelLatch;
@@ -123,7 +121,7 @@ public class TaskTest extends TestLogger {
private BlockingQueue<Object> taskManagerMessages;
private BlockingQueue<Object> jobManagerMessages;
private BlockingQueue<Object> listenerMessages;
-
+
@Before
public void createQueuesAndActors() {
taskManagerMessages = new LinkedBlockingQueue<>();
@@ -135,7 +133,7 @@ public class TaskTest extends TestLogger {
listener = new ActorGatewayTaskExecutionStateListener(listenerGateway);
taskManagerConnection = new ActorGatewayTaskManagerActions(taskManagerGateway);
-
+
awaitLatch = new OneShotLatch();
triggerLatch = new OneShotLatch();
cancelLatch = new OneShotLatch();
@@ -155,32 +153,32 @@ public class TaskTest extends TestLogger {
// ------------------------------------------------------------------------
// Tests
// ------------------------------------------------------------------------
-
+
@Test
public void testRegularExecution() {
try {
Task task = createTask(TestInvokableCorrect.class);
-
+
// task should be new and perfect
assertEquals(ExecutionState.CREATED, task.getExecutionState());
assertFalse(task.isCanceledOrFailed());
assertNull(task.getFailureCause());
-
+
task.registerExecutionListener(listener);
-
+
// go into the run method. we should switch to DEPLOYING, RUNNING, then
// FINISHED, and all should be good
task.run();
-
+
// verify final state
assertEquals(ExecutionState.FINISHED, task.getExecutionState());
assertFalse(task.isCanceledOrFailed());
assertNull(task.getFailureCause());
-
+
// verify listener messages
validateListenerMessage(ExecutionState.RUNNING, task, false);
validateListenerMessage(ExecutionState.FINISHED, task, false);
-
+
// make sure that the TaskManager received an message to unregister the task
validateTaskManagerStateChange(ExecutionState.RUNNING, task, false);
validateUnregisterTask(task.getExecutionId());
@@ -198,7 +196,7 @@ public class TaskTest extends TestLogger {
task.cancelExecution();
assertEquals(ExecutionState.CANCELING, task.getExecutionState());
-
+
task.run();
// verify final state
@@ -274,7 +272,7 @@ public class TaskTest extends TestLogger {
// mock a working library cache
LibraryCacheManager libCache = mock(LibraryCacheManager.class);
when(libCache.getClassLoader(any(JobID.class))).thenReturn(getClass().getClassLoader());
-
+
// mock a network manager that rejects registration
ResultPartitionManager partitionManager = mock(ResultPartitionManager.class);
ResultPartitionConsumableNotifier consumableNotifier = mock(ResultPartitionConsumableNotifier.class);
@@ -296,7 +294,7 @@ public class TaskTest extends TestLogger {
assertEquals(ExecutionState.FAILED, task.getExecutionState());
assertTrue(task.isCanceledOrFailed());
assertTrue(task.getFailureCause().getMessage().contains("buffers"));
-
+
validateUnregisterTask(task.getExecutionId());
validateListenerMessage(ExecutionState.FAILED, task, true);
}
@@ -326,13 +324,13 @@ public class TaskTest extends TestLogger {
fail(e.getMessage());
}
}
-
+
@Test
public void testExecutionFailsInInvoke() {
try {
Task task = createTask(InvokableWithExceptionInInvoke.class);
task.registerExecutionListener(listener);
-
+
task.run();
assertEquals(ExecutionState.FAILED, task.getExecutionState());
@@ -343,7 +341,7 @@ public class TaskTest extends TestLogger {
validateTaskManagerStateChange(ExecutionState.RUNNING, task, false);
validateUnregisterTask(task.getExecutionId());
-
+
validateListenerMessage(ExecutionState.RUNNING, task, false);
validateListenerMessage(ExecutionState.FAILED, task, true);
}
@@ -378,7 +376,7 @@ public class TaskTest extends TestLogger {
fail(e.getMessage());
}
}
-
+
@Test
public void testCancelDuringInvoke() {
try {
@@ -403,7 +401,7 @@ public class TaskTest extends TestLogger {
validateTaskManagerStateChange(ExecutionState.RUNNING, task, false);
validateUnregisterTask(task.getExecutionId());
-
+
validateListenerMessage(ExecutionState.RUNNING, task, false);
validateCancelingAndCanceledListenerMessage(task);
}
@@ -463,7 +461,7 @@ public class TaskTest extends TestLogger {
validateTaskManagerStateChange(ExecutionState.RUNNING, task, false);
validateUnregisterTask(task.getExecutionId());
-
+
validateListenerMessage(ExecutionState.RUNNING, task, false);
validateListenerMessage(ExecutionState.FAILED, task, true);
}
@@ -472,7 +470,7 @@ public class TaskTest extends TestLogger {
fail(e.getMessage());
}
}
-
+
@Test
public void testExecutionFailesAfterCanceling() {
try {
@@ -487,7 +485,7 @@ public class TaskTest extends TestLogger {
task.cancelExecution();
assertEquals(ExecutionState.CANCELING, task.getExecutionState());
-
+
// this causes an exception
triggerLatch.trigger();
@@ -497,7 +495,7 @@ public class TaskTest extends TestLogger {
assertEquals(ExecutionState.CANCELED, task.getExecutionState());
assertTrue(task.isCanceledOrFailed());
assertNull(task.getFailureCause());
-
+
validateTaskManagerStateChange(ExecutionState.RUNNING, task, false);
validateUnregisterTask(task.getExecutionId());
@@ -529,11 +527,11 @@ public class TaskTest extends TestLogger {
triggerLatch.trigger();
task.getExecutingThread().join();
-
+
assertEquals(ExecutionState.FAILED, task.getExecutionState());
assertTrue(task.isCanceledOrFailed());
assertTrue(task.getFailureCause().getMessage().contains("external"));
-
+
validateTaskManagerStateChange(ExecutionState.RUNNING, task, false);
validateUnregisterTask(task.getExecutionId());
@@ -606,7 +604,7 @@ public class TaskTest extends TestLogger {
expected.put(ExecutionState.SCHEDULED, ExecutionState.RUNNING);
expected.put(ExecutionState.DEPLOYING, ExecutionState.RUNNING);
expected.put(ExecutionState.FINISHED, ExecutionState.RUNNING);
-
+
expected.put(ExecutionState.CANCELED, ExecutionState.CANCELING);
expected.put(ExecutionState.CANCELING, ExecutionState.CANCELING);
expected.put(ExecutionState.FAILED, ExecutionState.CANCELING);
@@ -901,10 +899,9 @@ public class TaskTest extends TestLogger {
* @return BlobCache mock with the bare minimum of implemented functions that work
*/
private BlobCacheService createBlobCache() {
- BlobCacheService blobService =
- new BlobCacheService(mock(PermanentBlobCache.class), mock(TransientBlobCache.class));
-
- return blobService;
+ return new BlobCacheService(
+ mock(PermanentBlobCache.class),
+ mock(TransientBlobCache.class));
}
private Task createTask(Class<? extends AbstractInvokable> invokable) throws IOException {
@@ -915,14 +912,14 @@ public class TaskTest extends TestLogger {
BlobCacheService blobService = createBlobCache();
LibraryCacheManager libCache = mock(LibraryCacheManager.class);
when(libCache.getClassLoader(any(JobID.class))).thenReturn(getClass().getClassLoader());
- return createTask(invokable, blobService,libCache, config, new ExecutionConfig());
+ return createTask(invokable, blobService, libCache, config, new ExecutionConfig());
}
private Task createTask(Class<? extends AbstractInvokable> invokable, Configuration config, ExecutionConfig execConfig) throws IOException {
BlobCacheService blobService = createBlobCache();
LibraryCacheManager libCache = mock(LibraryCacheManager.class);
when(libCache.getClassLoader(any(JobID.class))).thenReturn(getClass().getClassLoader());
- return createTask(invokable, blobService,libCache, config, execConfig);
+ return createTask(invokable, blobService, libCache, config, execConfig);
}
private Task createTask(
@@ -930,7 +927,7 @@ public class TaskTest extends TestLogger {
BlobCacheService blobService,
LibraryCacheManager libCache) throws IOException {
- return createTask(invokable, blobService,libCache, new Configuration(), new ExecutionConfig());
+ return createTask(invokable, blobService, libCache, new Configuration(), new ExecutionConfig());
}
private Task createTask(
@@ -965,7 +962,7 @@ public class TaskTest extends TestLogger {
Executor executor) throws IOException {
return createTask(invokable, blobService, libCache, networkEnvironment, consumableNotifier, partitionProducerStateChecker, executor, new Configuration(), new ExecutionConfig());
}
-
+
private Task createTask(
Class<? extends AbstractInvokable> invokable,
BlobCacheService blobService,
@@ -976,7 +973,7 @@ public class TaskTest extends TestLogger {
Executor executor,
Configuration taskManagerConfig,
ExecutionConfig execConfig) throws IOException {
-
+
JobID jobId = new JobID();
JobVertexID jobVertexId = new JobVertexID();
ExecutionAttemptID executionAttemptId = new ExecutionAttemptID();
@@ -1010,7 +1007,7 @@ public class TaskTest extends TestLogger {
TaskMetricGroup taskMetricGroup = mock(TaskMetricGroup.class);
when(taskMetricGroup.getIOMetricGroup()).thenReturn(mock(TaskIOMetricGroup.class));
-
+
return new Task(
jobInformation,
taskInformation,
@@ -1018,8 +1015,8 @@ public class TaskTest extends TestLogger {
new AllocationID(),
0,
0,
- Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
- Collections.<InputGateDeploymentDescriptor>emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
0,
mock(MemoryManager.class),
mock(IOManager.class),
@@ -1042,18 +1039,18 @@ public class TaskTest extends TestLogger {
// ------------------------------------------------------------------------
// Validation Methods
// ------------------------------------------------------------------------
-
+
private void validateUnregisterTask(ExecutionAttemptID id) {
try {
// we may have to wait for a bit to give the actors time to receive the message
// and put it into the queue
Object rawMessage = taskManagerMessages.take();
-
+
assertNotNull("There is no additional TaskManager message", rawMessage);
if (!(rawMessage instanceof TaskMessages.TaskInFinalState)) {
fail("TaskManager message is not 'UnregisterTask', but " + rawMessage.getClass());
}
-
+
TaskMessages.TaskInFinalState message = (TaskMessages.TaskInFinalState) rawMessage;
assertEquals(id, message.executionID());
}
@@ -1072,10 +1069,10 @@ public class TaskTest extends TestLogger {
if (!(rawMessage instanceof TaskMessages.UpdateTaskExecutionState)) {
fail("TaskManager message is not 'UpdateTaskExecutionState', but " + rawMessage.getClass());
}
-
+
TaskMessages.UpdateTaskExecutionState message =
(TaskMessages.UpdateTaskExecutionState) rawMessage;
-
+
TaskExecutionState taskState = message.taskExecutionState();
assertEquals(task.getJobID(), taskState.getJobID());
@@ -1092,7 +1089,7 @@ public class TaskTest extends TestLogger {
fail("interrupted");
}
}
-
+
private void validateListenerMessage(ExecutionState state, Task task, boolean hasError) {
try {
// we may have to wait for a bit to give the actors time to receive the message
@@ -1100,13 +1097,13 @@ public class TaskTest extends TestLogger {
TaskMessages.UpdateTaskExecutionState message =
(TaskMessages.UpdateTaskExecutionState) listenerMessages.take();
assertNotNull("There is no additional listener message", message);
-
+
TaskExecutionState taskState = message.taskExecutionState();
assertEquals(task.getJobID(), taskState.getJobID());
assertEquals(task.getExecutionId(), taskState.getID());
assertEquals(state, taskState.getExecutionState());
-
+
if (hasError) {
assertNotNull(taskState.getError(getClass().getClassLoader()));
} else {
@@ -1126,8 +1123,7 @@ public class TaskTest extends TestLogger {
(TaskMessages.UpdateTaskExecutionState) listenerMessages.take();
TaskMessages.UpdateTaskExecutionState message2 =
(TaskMessages.UpdateTaskExecutionState) listenerMessages.take();
-
-
+
assertNotNull("There is no additional listener message", message1);
assertNotNull("There is no additional listener message", message2);
@@ -1138,30 +1134,31 @@ public class TaskTest extends TestLogger {
assertEquals(task.getJobID(), taskState2.getJobID());
assertEquals(task.getExecutionId(), taskState1.getID());
assertEquals(task.getExecutionId(), taskState2.getID());
-
+
ExecutionState state1 = taskState1.getExecutionState();
ExecutionState state2 = taskState2.getExecutionState();
-
+
// it may be (very rarely) that the following race happens:
// - OUTSIDE THREAD: call to cancel()
// - OUTSIDE THREAD: atomic state change from running to canceling
// - TASK THREAD: finishes, atomic change from canceling to canceled
// - TASK THREAD: send notification that state is canceled
// - OUTSIDE THREAD: send notification that state is canceling
-
+
// for that reason, we allow the notification messages in any order.
- assertTrue( (state1 == ExecutionState.CANCELING && state2 == ExecutionState.CANCELED) ||
+ assertTrue((state1 == ExecutionState.CANCELING && state2 == ExecutionState.CANCELED) ||
(state2 == ExecutionState.CANCELING && state1 == ExecutionState.CANCELED));
}
catch (InterruptedException e) {
fail("interrupted");
}
}
-
+
// --------------------------------------------------------------------------------------------
// Mock invokable code
// --------------------------------------------------------------------------------------------
-
+
+ /** Test task class. */
public static final class TestInvokableCorrect extends AbstractInvokable {
public TestInvokableCorrect(Environment environment) {
@@ -1177,6 +1174,7 @@ public class TaskTest extends TestLogger {
}
}
+ /** Test task class. */
public static final class InvokableWithExceptionInInvoke extends AbstractInvokable {
public InvokableWithExceptionInInvoke(Environment environment) {
@@ -1189,6 +1187,7 @@ public class TaskTest extends TestLogger {
}
}
+ /** Test task class. */
public static final class InvokableWithExceptionOnTrigger extends AbstractInvokable {
public InvokableWithExceptionOnTrigger(Environment environment) {
@@ -1198,7 +1197,7 @@ public class TaskTest extends TestLogger {
@Override
public void invoke() {
awaitLatch.trigger();
-
+
// make sure that the interrupt call does not
// grab us out of the lock early
while (true) {
@@ -1215,13 +1214,15 @@ public class TaskTest extends TestLogger {
}
}
- public static abstract class InvokableNonInstantiable extends AbstractInvokable {
+ /** Test task class. */
+ public abstract static class InvokableNonInstantiable extends AbstractInvokable {
public InvokableNonInstantiable(Environment environment) {
super(environment);
}
}
+ /** Test task class. */
public static final class InvokableBlockingInInvoke extends AbstractInvokable {
public InvokableBlockingInInvoke(Environment environment) {
@@ -1231,7 +1232,7 @@ public class TaskTest extends TestLogger {
@Override
public void invoke() throws Exception {
awaitLatch.trigger();
-
+
// block forever
synchronized (this) {
wait();
@@ -1239,6 +1240,7 @@ public class TaskTest extends TestLogger {
}
}
+ /** Test task class. */
public static final class InvokableWithCancelTaskExceptionInInvoke extends AbstractInvokable {
public InvokableWithCancelTaskExceptionInInvoke(Environment environment) {
@@ -1253,11 +1255,12 @@ public class TaskTest extends TestLogger {
triggerLatch.await();
}
catch (Throwable ignored) {}
-
+
throw new CancelTaskException();
}
}
+ /** Test task class. */
public static final class InvokableInterruptableSharedLockInInvokeAndCancel extends AbstractInvokable {
private final Object lock = new Object();
@@ -1282,6 +1285,7 @@ public class TaskTest extends TestLogger {
}
}
+ /** Test task class. */
public static final class InvokableBlockingInCancel extends AbstractInvokable {
public InvokableBlockingInCancel(Environment environment) {
@@ -1313,6 +1317,7 @@ public class TaskTest extends TestLogger {
}
}
+ /** Test task class. */
public static final class InvokableUninterruptibleBlockingInvoke extends AbstractInvokable {
public InvokableUninterruptibleBlockingInvoke(Environment environment) {
@@ -1337,6 +1342,7 @@ public class TaskTest extends TestLogger {
}
}
+ /** Test task class. */
public static final class FailingInvokableWithChainedException extends AbstractInvokable {
public FailingInvokableWithChainedException(Environment environment) {