This is an automated email from the ASF dual-hosted git repository.
RocMarshal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new a3c59ecec4b [FLINK-40960][runtime][JUnit5 Migration] Migrate the
remaining of testing classes in taskmanager package of flink-runtime module
a3c59ecec4b is described below
commit a3c59ecec4be7cd406318103dd7691a4d78a4829
Author: Yuepeng Pan <[email protected]>
AuthorDate: Thu Oct 8 23:00:50 2026 +0800
[FLINK-40960][runtime][JUnit5 Migration] Migrate the remaining of testing
classes in taskmanager package of flink-runtime module
Co-authored-by: Sergey Nuyanzin <[email protected]>
---
.../runtime/taskmanager/TaskAsyncCallTest.java | 36 +-
.../taskmanager/TaskExecutionStateTest.java | 30 +-
.../apache/flink/runtime/taskmanager/TaskTest.java | 395 ++++++++++-----------
3 files changed, 224 insertions(+), 237 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
index 254165d22f9..7da1f089e66 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
@@ -58,11 +58,12 @@ import
org.apache.flink.runtime.taskexecutor.TestGlobalAggregateManager;
import org.apache.flink.runtime.util.TestingTaskManagerRuntimeInfo;
import org.apache.flink.runtime.util.TestingUserCodeClassLoader;
import org.apache.flink.util.SerializedValue;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
@@ -70,13 +71,12 @@ import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import static
org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
-import static org.hamcrest.Matchers.isOneOf;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
/** Testing asynchronous call of {@link Task}. */
-public class TaskAsyncCallTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class TaskAsyncCallTest {
/** Number of expected checkpoints. */
private static int numCalls;
@@ -92,8 +92,8 @@ public class TaskAsyncCallTest extends TestLogger {
private ShuffleEnvironment<?, ?> shuffleEnvironment;
- @Before
- public void createQueuesAndActors() {
+ @BeforeEach
+ void createQueuesAndActors() {
numCalls = 1000;
awaitLatch = new OneShotLatch();
@@ -102,8 +102,8 @@ public class TaskAsyncCallTest extends TestLogger {
shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
}
- @After
- public void teardown() throws Exception {
+ @AfterEach
+ void teardown() throws Exception {
if (shuffleEnvironment != null) {
shuffleEnvironment.close();
}
@@ -114,7 +114,7 @@ public class TaskAsyncCallTest extends TestLogger {
// ------------------------------------------------------------------------
@Test
- public void testCheckpointCallsInOrder() throws Exception {
+ void testCheckpointCallsInOrder() throws Exception {
Task task = createTask(CheckpointsInOrderInvokable.class);
try (TaskCleaner ignored = new TaskCleaner(task)) {
@@ -129,15 +129,15 @@ public class TaskAsyncCallTest extends TestLogger {
triggerLatch.await();
- assertFalse(task.isCanceledOrFailed());
+ assertThat(task.isCanceledOrFailed()).isFalse();
ExecutionState currentState = task.getExecutionState();
- assertThat(currentState, isOneOf(ExecutionState.RUNNING,
ExecutionState.FINISHED));
+ assertThat(currentState).isIn(ExecutionState.RUNNING,
ExecutionState.FINISHED);
}
}
@Test
- public void testMixedAsyncCallsInOrder() throws Exception {
+ void testMixedAsyncCallsInOrder() throws Exception {
Task task = createTask(CheckpointsInOrderInvokable.class);
try (TaskCleaner ignored = new TaskCleaner(task)) {
@@ -153,10 +153,10 @@ public class TaskAsyncCallTest extends TestLogger {
triggerLatch.await();
- assertFalse(task.isCanceledOrFailed());
+ assertThat(task.isCanceledOrFailed()).isFalse();
ExecutionState currentState = task.getExecutionState();
- assertThat(currentState, isOneOf(ExecutionState.RUNNING,
ExecutionState.FINISHED));
+ assertThat(currentState).isIn(ExecutionState.RUNNING,
ExecutionState.FINISHED);
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskExecutionStateTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskExecutionStateTest.java
index c095164b1ea..4c7dfbf8d70 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskExecutionStateTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskExecutionStateTest.java
@@ -22,24 +22,24 @@ import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.io.PrintStream;
import java.io.PrintWriter;
import static
org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
/**
* Correctness tests for hash/equals and serialization for the {@link
* org.apache.flink.runtime.taskmanager.TaskExecutionState}.
*/
-public class TaskExecutionStateTest {
+class TaskExecutionStateTest {
@Test
- public void testEqualsHashCode() {
+ void testEqualsHashCode() {
try {
final ExecutionAttemptID executionId = createExecutionAttemptId();
final ExecutionState state = ExecutionState.RUNNING;
@@ -48,8 +48,8 @@ public class TaskExecutionStateTest {
TaskExecutionState s1 = new TaskExecutionState(executionId, state,
error);
TaskExecutionState s2 = new TaskExecutionState(executionId, state,
error);
- assertEquals(s1.hashCode(), s2.hashCode());
- assertEquals(s1, s2);
+ assertThat(s1).hasSameHashCodeAs(s2);
+ assertThat(s1).isEqualTo(s2);
} catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
@@ -57,7 +57,7 @@ public class TaskExecutionStateTest {
}
@Test
- public void testSerialization() {
+ void testSerialization() {
try {
final ExecutionAttemptID executionId = createExecutionAttemptId();
final ExecutionState state = ExecutionState.DEPLOYING;
@@ -70,15 +70,15 @@ public class TaskExecutionStateTest {
TaskExecutionState javaSerCopy2 =
CommonTestUtils.createCopySerializable(original2);
// equalities
- assertEquals(original1, javaSerCopy1);
- assertEquals(javaSerCopy1, original1);
+ assertThat(javaSerCopy1).isEqualTo(original1);
+ assertThat(original1).isEqualTo(javaSerCopy1);
- assertEquals(original2, javaSerCopy2);
- assertEquals(javaSerCopy2, original2);
+ assertThat(javaSerCopy2).isEqualTo(original2);
+ assertThat(original2).isEqualTo(javaSerCopy2);
// hash codes
- assertEquals(original1.hashCode(), javaSerCopy1.hashCode());
- assertEquals(original2.hashCode(), javaSerCopy2.hashCode());
+ assertThat(javaSerCopy1).hasSameHashCodeAs(original1);
+ assertThat(javaSerCopy2).hasSameHashCodeAs(original2);
} catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
@@ -86,7 +86,7 @@ public class TaskExecutionStateTest {
}
@Test
- public void handleNonSerializableException() {
+ void handleNonSerializableException() {
try {
@SuppressWarnings({"ThrowableInstanceNeverThrown", "serial"})
Exception hostile =
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 36a248934e1..3b326ba2946 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
@@ -51,18 +51,18 @@ import
org.apache.flink.runtime.state.CheckpointStorageLocationReference;
import org.apache.flink.runtime.taskexecutor.PartitionProducerStateChecker;
import org.apache.flink.runtime.util.NettyShuffleDescriptorBuilder;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
import org.apache.flink.util.WrappingRuntimeException;
import org.apache.flink.util.concurrent.Executors;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
import javax.annotation.Nonnull;
@@ -82,17 +82,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import static
org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition;
-import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.CoreMatchers.instanceOf;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.CoreMatchers.notNullValue;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -103,43 +94,42 @@ import static org.mockito.Mockito.when;
* Tests for the Task, which make sure that correct state transitions happen,
and failures are
* correctly handled.
*/
-public class TaskTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class TaskTest {
private static final String RESTORE_EXCEPTION_MSG =
"TestExceptionInRestore";
private ShuffleEnvironment<?, ?> shuffleEnvironment;
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService>
EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
-
- @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new
TemporaryFolder();
+ @RegisterExtension
+ private static final TestExecutorExtension<ScheduledExecutorService>
EXECUTOR_RESOURCE =
+ TestingUtils.defaultExecutorExtension();
private static boolean wasCleanedUp = false;
- @Before
- public void setup() {
+ @BeforeEach
+ void setup() {
shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
wasCleanedUp = false;
}
- @After
- public void teardown() throws Exception {
+ @AfterEach
+ void teardown() throws Exception {
if (shuffleEnvironment != null) {
shuffleEnvironment.close();
}
}
@Test
- public void testTaskFailedWithCleanupCancelledExternally() throws
Exception {
+ void testTaskFailedWithCleanupCancelledExternally() throws Exception {
testTaskFailedWithCleanupFailingExternally(false);
}
@Test
- public void testTaskFailedWithCleanupFailingExternally() throws Exception {
+ void testTaskFailedWithCleanupFailingExternally() throws Exception {
testTaskFailedWithCleanupFailingExternally(true);
}
- public void testTaskFailedWithCleanupFailingExternally(boolean cancelled)
throws Exception {
+ void testTaskFailedWithCleanupFailingExternally(boolean cancelled) throws
Exception {
Task task =
createTaskBuilder()
.setInvokable(
@@ -150,30 +140,33 @@ public class TaskTest extends TestLogger {
.build(Executors.directExecutor());
task.run();
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- ExceptionUtils.assertThrowable(task.getFailureCause(),
ExpectedTestException.class);
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(
+ ExceptionUtils.findThrowable(
+ task.getFailureCause(),
ExpectedTestException.class))
+ .isPresent();
}
@Test
- public void testCleanupWhenRestoreFails() throws Exception {
+ void testCleanupWhenRestoreFails() throws Exception {
createTaskBuilder()
.setInvokable(InvokableWithExceptionInRestore.class)
.build(Executors.directExecutor())
.run();
- assertTrue(wasCleanedUp);
+ assertThat(wasCleanedUp).isTrue();
}
@Test
- public void testCleanupWhenInvokeFails() throws Exception {
+ void testCleanupWhenInvokeFails() throws Exception {
createTaskBuilder()
.setInvokable(InvokableWithExceptionInInvoke.class)
.build(Executors.directExecutor())
.run();
- assertTrue(wasCleanedUp);
+ assertThat(wasCleanedUp).isTrue();
}
@Test
- public void testCleanupWhenCancelledAfterRestore() throws Exception {
+ void testCleanupWhenCancelledAfterRestore() throws Exception {
Task task =
createTaskBuilder()
.setInvokable(InvokableBlockingInRestore.class)
@@ -182,22 +175,22 @@ public class TaskTest extends TestLogger {
awaitInvokableLatch(task);
task.cancelExecution();
task.getExecutingThread().join();
- assertTrue(wasCleanedUp);
+ assertThat(wasCleanedUp).isTrue();
}
@Test
- public void testCleanupWhenAfterInvokeSucceeded() throws Exception {
+ void testCleanupWhenAfterInvokeSucceeded() throws Exception {
Task task =
createTaskBuilder()
.setInvokable(TestInvokableCorrect.class)
.build(Executors.directExecutor());
task.run();
- assertTrue(wasCleanedUp);
- assertFalse(task.isCanceledOrFailed());
+ assertThat(wasCleanedUp).isTrue();
+ assertThat(task.isCanceledOrFailed()).isFalse();
}
@Test
- public void testCleanupWhenSwitchToInitializationFails() throws Exception {
+ void testCleanupWhenSwitchToInitializationFails() throws Exception {
Task task =
createTaskBuilder()
.setInvokable(TestInvokableCorrect.class)
@@ -214,12 +207,12 @@ public class TaskTest extends TestLogger {
})
.build(Executors.directExecutor());
task.run();
- assertTrue(wasCleanedUp);
- assertTrue(task.isCanceledOrFailed());
+ assertThat(wasCleanedUp).isTrue();
+ assertThat(task.isCanceledOrFailed()).isTrue();
}
@Test
- public void testRegularExecution() throws Exception {
+ void testRegularExecution() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -228,19 +221,19 @@ public class TaskTest extends TestLogger {
.build(Executors.directExecutor());
// task should be new and perfect
- assertEquals(ExecutionState.CREATED, task.getExecutionState());
- assertFalse(task.isCanceledOrFailed());
- assertNull(task.getFailureCause());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.CREATED);
+ assertThat(task.isCanceledOrFailed()).isFalse();
+ assertThat(task.getFailureCause()).isNull();
// 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());
- assertNull(task.getInvokable());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FINISHED);
+ assertThat(task.isCanceledOrFailed()).isFalse();
+ assertThat(task.getFailureCause()).isNull();
+ assertThat(task.getInvokable()).isNull();
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING,
task, null);
@@ -248,35 +241,35 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testCancelRightAway() throws Exception {
+ void testCancelRightAway() throws Exception {
final Task task =
createTaskBuilder().build(Executors.directExecutor());
task.cancelExecution();
- assertEquals(ExecutionState.CANCELING, task.getExecutionState());
+
assertThat(task.getExecutionState()).isSameAs(ExecutionState.CANCELING);
task.run();
// verify final state
- assertEquals(ExecutionState.CANCELED, task.getExecutionState());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.CANCELED);
- assertNull(task.getInvokable());
+ assertThat(task.getInvokable()).isNull();
}
@Test
- public void testFailExternallyRightAway() throws Exception {
+ void testFailExternallyRightAway() throws Exception {
final Task task =
createTaskBuilder().build(Executors.directExecutor());
task.failExternally(new Exception("fail externally"));
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
task.run();
// verify final state
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
}
@Test
- public void testLibraryCacheRegistrationFailed() throws Exception {
+ void testLibraryCacheRegistrationFailed() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final IOException testException = new IOException("Could not load
classloader");
final Task task =
@@ -292,25 +285,25 @@ public class TaskTest extends TestLogger {
.build(Executors.directExecutor());
// task should be new and perfect
- assertEquals(ExecutionState.CREATED, task.getExecutionState());
- assertFalse(task.isCanceledOrFailed());
- assertNull(task.getFailureCause());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.CREATED);
+ assertThat(task.isCanceledOrFailed()).isFalse();
+ assertThat(task.getFailureCause()).isNull();
// should fail
task.run();
// verify final state
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertThat(task.getFailureCause(), is(testException));
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).isSameAs(testException);
- assertNull(task.getInvokable());
+ assertThat(task.getInvokable()).isNull();
taskManagerActions.validateListenerMessage(ExecutionState.FAILED,
task, testException);
}
@Test
- public void testExecutionFailsInNetworkRegistrationForPartitions() throws
Exception {
+ void testExecutionFailsInNetworkRegistrationForPartitions() throws
Exception {
final PartitionDescriptor partitionDescriptor =
PartitionDescriptorBuilder.newBuilder().build();
final ShuffleDescriptor shuffleDescriptor =
@@ -322,7 +315,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testExecutionFailsInNetworkRegistrationForGates() throws
Exception {
+ void testExecutionFailsInNetworkRegistrationForGates() throws Exception {
final ShuffleDescriptor dummyChannel =
NettyShuffleDescriptorBuilder.newBuilder().buildRemote();
final InputGateDeploymentDescriptor dummyGate =
@@ -362,16 +355,16 @@ public class TaskTest extends TestLogger {
task.run();
// verify final state
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertTrue(task.getFailureCause().getMessage().contains(errorMessage));
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).hasMessageContaining(errorMessage);
taskManagerActions.validateListenerMessage(
ExecutionState.FAILED, task, new
IllegalStateException(errorMessage));
}
@Test
- public void testInvokableInstantiationFailed() throws Exception {
+ void testInvokableInstantiationFailed() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -383,9 +376,9 @@ public class TaskTest extends TestLogger {
task.run();
// verify final state
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
-
assertTrue(task.getFailureCause().getMessage().contains("instantiate"));
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).hasMessageContaining("instantiate");
taskManagerActions.validateListenerMessage(
ExecutionState.FAILED,
@@ -394,7 +387,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testExecutionFailsInRestore() throws Exception {
+ void testExecutionFailsInRestore() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -404,11 +397,10 @@ public class TaskTest extends TestLogger {
task.run();
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertNotNull(task.getFailureCause());
- assertNotNull(task.getFailureCause().getMessage());
- assertThat(task.getFailureCause().getMessage(),
containsString(RESTORE_EXCEPTION_MSG));
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).isNotNull();
+
assertThat(task.getFailureCause()).hasMessageContaining(RESTORE_EXCEPTION_MSG);
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(
@@ -416,7 +408,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testExecutionFailsInInvoke() throws Exception {
+ void testExecutionFailsInInvoke() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -426,11 +418,10 @@ public class TaskTest extends TestLogger {
task.run();
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertNotNull(task.getFailureCause());
- assertNotNull(task.getFailureCause().getMessage());
- assertTrue(task.getFailureCause().getMessage().contains("test"));
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).isNotNull();
+ assertThat(task.getFailureCause()).hasMessageContaining("test");
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING,
task, null);
@@ -439,7 +430,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testFailWithWrappedException() throws Exception {
+ void testFailWithWrappedException() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -449,11 +440,11 @@ public class TaskTest extends TestLogger {
task.run();
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
final Throwable cause = task.getFailureCause();
- assertTrue(cause instanceof IOException);
+ assertThat(cause).isInstanceOf(IOException.class);
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING,
task, null);
@@ -462,7 +453,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testCancelDuringRestore() throws Exception {
+ void testCancelDuringRestore() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -477,23 +468,21 @@ public class TaskTest extends TestLogger {
awaitInvokableLatch(task);
task.cancelExecution();
- assertTrue(
- task.getExecutionState().toString(),
- task.getExecutionState() == ExecutionState.CANCELING
- || task.getExecutionState() ==
ExecutionState.CANCELED);
+ assertThat(task.getExecutionState())
+ .isIn(ExecutionState.CANCELING, ExecutionState.CANCELED);
task.getExecutingThread().join();
- assertEquals(ExecutionState.CANCELED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertNull(task.getFailureCause());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.CANCELED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).isNull();
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(ExecutionState.CANCELED,
task, null);
}
@Test
- public void testCancelDuringInvoke() throws Exception {
+ void testCancelDuringInvoke() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -508,16 +497,14 @@ public class TaskTest extends TestLogger {
awaitInvokableLatch(task);
task.cancelExecution();
- assertTrue(
- task.getExecutionState().toString(),
- task.getExecutionState() == ExecutionState.CANCELING
- || task.getExecutionState() ==
ExecutionState.CANCELED);
+ assertThat(task.getExecutionState())
+ .isIn(ExecutionState.CANCELING, ExecutionState.CANCELED);
task.getExecutingThread().join();
- assertEquals(ExecutionState.CANCELED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertNull(task.getFailureCause());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.CANCELED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).isNull();
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING,
task, null);
@@ -525,7 +512,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testFailExternallyDuringRestore() throws Exception {
+ void testFailExternallyDuringRestore() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -543,9 +530,9 @@ public class TaskTest extends TestLogger {
task.getExecutingThread().join();
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertThat(task.getFailureCause().getMessage(),
containsString(RESTORE_EXCEPTION_MSG));
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+
assertThat(task.getFailureCause()).hasMessageContaining(RESTORE_EXCEPTION_MSG);
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(
@@ -553,7 +540,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testFailExternallyDuringInvoke() throws Exception {
+ void testFailExternallyDuringInvoke() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -571,9 +558,9 @@ public class TaskTest extends TestLogger {
task.getExecutingThread().join();
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertTrue(task.getFailureCause().getMessage().contains("test"));
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).hasMessageContaining("test");
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING,
task, null);
@@ -582,7 +569,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testCanceledAfterExecutionFailedInInvoke() throws Exception {
+ void testCanceledAfterExecutionFailedInInvoke() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -595,9 +582,9 @@ public class TaskTest extends TestLogger {
// this should not overwrite the failure state
task.cancelExecution();
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertTrue(task.getFailureCause().getMessage().contains("test"));
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).hasMessageContaining("test");
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING,
task, null);
@@ -606,7 +593,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testExecutionFailsAfterCanceling() throws Exception {
+ void testExecutionFailsAfterCanceling() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -621,7 +608,7 @@ public class TaskTest extends TestLogger {
awaitInvokableLatch(task);
task.cancelExecution();
- assertEquals(ExecutionState.CANCELING, task.getExecutionState());
+
assertThat(task.getExecutionState()).isSameAs(ExecutionState.CANCELING);
// this causes an exception
triggerInvokableLatch(task);
@@ -629,9 +616,9 @@ public class TaskTest extends TestLogger {
task.getExecutingThread().join();
// we should still be in state canceled
- assertEquals(ExecutionState.CANCELED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertNull(task.getFailureCause());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.CANCELED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).isNull();
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING,
task, null);
@@ -639,7 +626,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testExecutionFailsAfterTaskMarkedFailed() throws Exception {
+ void testExecutionFailsAfterTaskMarkedFailed() throws Exception {
final QueuedNoOpTaskManagerActions taskManagerActions = new
QueuedNoOpTaskManagerActions();
final Task task =
createTaskBuilder()
@@ -654,16 +641,16 @@ public class TaskTest extends TestLogger {
awaitInvokableLatch(task);
task.failExternally(new Exception("external"));
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
// this causes an exception
triggerInvokableLatch(task);
task.getExecutingThread().join();
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertTrue(task.getFailureCause().getMessage().contains("external"));
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).hasMessageContaining("external");
taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task,
null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING,
task, null);
@@ -672,7 +659,7 @@ public class TaskTest extends TestLogger {
}
@Test
- public void testCancelTaskException() throws Exception {
+ void testCancelTaskException() throws Exception {
final Task task =
createTaskBuilder()
.setInvokable(InvokableWithCancelTaskExceptionInInvoke.class)
@@ -684,11 +671,11 @@ public class TaskTest extends TestLogger {
triggerInvokableLatch(task);
task.getExecutingThread().join();
- assertEquals(ExecutionState.CANCELED, task.getExecutionState());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.CANCELED);
}
@Test
- public void testCancelTaskExceptionAfterTaskMarkedFailed() throws
Exception {
+ void testCancelTaskExceptionAfterTaskMarkedFailed() throws Exception {
final Task task =
createTaskBuilder()
.setInvokable(InvokableWithCancelTaskExceptionInInvoke.class)
@@ -700,7 +687,7 @@ public class TaskTest extends TestLogger {
awaitInvokableLatch(task);
task.failExternally(new Exception("external"));
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
// Either we cause the CancelTaskException or the TaskCanceler
// by interrupting the invokable.
@@ -708,13 +695,13 @@ public class TaskTest extends TestLogger {
task.getExecutingThread().join();
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
- assertTrue(task.isCanceledOrFailed());
- assertTrue(task.getFailureCause().getMessage().contains("external"));
+ assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
+ assertThat(task.isCanceledOrFailed()).isTrue();
+ assertThat(task.getFailureCause()).hasMessageContaining("external");
}
@Test
- public void testOnPartitionStateUpdateWhileRunning() throws Exception {
+ void testOnPartitionStateUpdateWhileRunning() throws Exception {
testOnPartitionStateUpdate(ExecutionState.RUNNING);
}
@@ -724,11 +711,11 @@ public class TaskTest extends TestLogger {
* input gates.
*/
@Test
- public void testOnPartitionStateUpdateWhileDeploying() throws Exception {
+ void testOnPartitionStateUpdateWhileDeploying() throws Exception {
testOnPartitionStateUpdate(ExecutionState.DEPLOYING);
}
- public void testOnPartitionStateUpdate(ExecutionState initialTaskState)
throws Exception {
+ void testOnPartitionStateUpdate(ExecutionState initialTaskState) throws
Exception {
final ResultPartitionID partitionId = new ResultPartitionID();
final Task task =
@@ -768,15 +755,15 @@ public class TaskTest extends TestLogger {
ExecutionState newTaskState = task.getExecutionState();
- assertEquals(expected.get(state), newTaskState);
+ assertThat(newTaskState).isSameAs(expected.get(state));
}
- assertEquals(5, producingStateCounter);
+ assertThat(producingStateCounter).isEqualTo(5);
}
/** Tests the trigger partition state update future completions. */
@Test
- public void testTriggerPartitionStateUpdate() throws Exception {
+ void testTriggerPartitionStateUpdate() throws Exception {
final IntermediateDataSetID resultId = new IntermediateDataSetID();
final ResultPartitionID partitionId = new ResultPartitionID();
@@ -811,12 +798,12 @@ public class TaskTest extends TestLogger {
partitionId,
checkResult ->
assertThat(
-
remoteChannelStateChecker.isProducerReadyOrAbortConsumption(
- checkResult),
- is(false)));
+ remoteChannelStateChecker
+
.isProducerReadyOrAbortConsumption(checkResult))
+ .isFalse());
promise.completeExceptionally(new
PartitionProducerDisposedException(partitionId));
- assertEquals(ExecutionState.CANCELING, task.getExecutionState());
+
assertThat(task.getExecutionState()).isSameAs(ExecutionState.CANCELING);
}
{
@@ -841,13 +828,13 @@ public class TaskTest extends TestLogger {
partitionId,
checkResult ->
assertThat(
-
remoteChannelStateChecker.isProducerReadyOrAbortConsumption(
- checkResult),
- is(false)));
+ remoteChannelStateChecker
+
.isProducerReadyOrAbortConsumption(checkResult))
+ .isFalse());
promise.completeExceptionally(new RuntimeException("Any other
exception"));
- assertEquals(ExecutionState.FAILED, task.getExecutionState());
+
assertThat(task.getExecutionState()).isSameAs(ExecutionState.FAILED);
}
{
@@ -885,9 +872,9 @@ public class TaskTest extends TestLogger {
promise.completeExceptionally(new TimeoutException());
- assertEquals(ExecutionState.RUNNING, task.getExecutionState());
+
assertThat(task.getExecutionState()).isSameAs(ExecutionState.RUNNING);
- assertEquals(1, callCount.get());
+ assertThat(callCount).hasValue(1);
} finally {
task.getExecutingThread().interrupt();
task.getExecutingThread().join();
@@ -928,9 +915,9 @@ public class TaskTest extends TestLogger {
promise.complete(ExecutionState.RUNNING);
- assertEquals(ExecutionState.RUNNING, task.getExecutionState());
+
assertThat(task.getExecutionState()).isSameAs(ExecutionState.RUNNING);
- assertEquals(1, callCount.get());
+ assertThat(callCount).hasValue(1);
} finally {
task.getExecutingThread().interrupt();
task.getExecutingThread().join();
@@ -943,7 +930,7 @@ public class TaskTest extends TestLogger {
* blocks the task canceller. Interrupt after cancel via cancellation
watch dog.
*/
@Test
- public void testWatchDogInterruptsTask() throws Exception {
+ void testWatchDogInterruptsTask() throws Exception {
final TaskManagerActions taskManagerActions = new
ProhibitFatalErrorTaskManagerActions();
final Configuration config = new Configuration();
@@ -970,7 +957,7 @@ public class TaskTest extends TestLogger {
* blocks the task canceller. Interrupt after cancel via cancellation
watch dog.
*/
@Test
- public void testWatchDogThrowFatalErrorOnTaskStuckInInstantiation() throws
Exception {
+ void testWatchDogThrowFatalErrorOnTaskStuckInInstantiation() throws
Exception {
final InterruptOnFatalErrorTaskManagerActions taskManagerActions =
new InterruptOnFatalErrorTaskManagerActions();
@@ -992,7 +979,7 @@ public class TaskTest extends TestLogger {
task.getExecutingThread().join();
// Expect fatal error to recover
- assertTrue(taskManagerActions.hasFatalError());
+ assertThat(taskManagerActions.hasFatalError()).isTrue();
}
/**
@@ -1001,7 +988,7 @@ public class TaskTest extends TestLogger {
* no fatal error.
*/
@Test
- public void testInterruptibleSharedLockInInvokeAndCancel() throws
Exception {
+ void testInterruptibleSharedLockInInvokeAndCancel() throws Exception {
final TaskManagerActions taskManagerActions = new
ProhibitFatalErrorTaskManagerActions();
final Configuration config = new Configuration();
@@ -1028,7 +1015,7 @@ public class TaskTest extends TestLogger {
* error.
*/
@Test
- public void testFatalErrorAfterUnInterruptibleInvoke() throws Exception {
+ void testFatalErrorAfterUnInterruptibleInvoke() throws Exception {
final CompletableFuture<Throwable> fatalErrorFuture = new
CompletableFuture<>();
final TestingTaskManagerActions taskManagerActions =
TestingTaskManagerActions.newBuilder()
@@ -1055,7 +1042,7 @@ public class TaskTest extends TestLogger {
// wait for the notification of notifyFatalError
final Throwable fatalError = fatalErrorFuture.join();
- assertThat(fatalError, is(notNullValue()));
+ assertThat(fatalError).isNotNull();
} finally {
// Interrupt again to clean up Thread
triggerInvokableLatch(task);
@@ -1066,7 +1053,7 @@ public class TaskTest extends TestLogger {
/** Tests that a fatal error gotten from canceling task is notified. */
@Test
- public void testFatalErrorOnCanceling() throws Exception {
+ void testFatalErrorOnCanceling() throws Exception {
final CompletableFuture<Throwable> fatalErrorFuture = new
CompletableFuture<>();
final TestingTaskManagerActions taskManagerActions =
TestingTaskManagerActions.newBuilder()
@@ -1102,7 +1089,7 @@ public class TaskTest extends TestLogger {
// wait for the notification of notifyFatalError
final Throwable fatalError = fatalErrorFuture.join();
- assertThat(fatalError, instanceOf(fatalErrorType));
+ assertThat(fatalError).isInstanceOf(fatalErrorType);
} finally {
triggerInvokableLatch(task);
}
@@ -1110,7 +1097,7 @@ public class TaskTest extends TestLogger {
/** Tests that the task configuration is respected and overwritten by the
execution config. */
@Test
- public void testTaskConfig() throws Exception {
+ void testTaskConfig() throws Exception {
long interval = 28218123;
long timeout = interval + 19292;
@@ -1129,24 +1116,24 @@ public class TaskTest extends TestLogger {
.setExecutionConfig(executionConfig)
.build(Executors.directExecutor());
- assertEquals(interval, task.getTaskCancellationInterval());
- assertEquals(timeout, task.getTaskCancellationTimeout());
+ assertThat(task.getTaskCancellationInterval()).isEqualTo(interval);
+ assertThat(task.getTaskCancellationTimeout()).isEqualTo(timeout);
task.startTaskThread();
awaitInvokableLatch(task);
- assertEquals(
- executionConfig.getTaskCancellationInterval(),
task.getTaskCancellationInterval());
- assertEquals(
- executionConfig.getTaskCancellationTimeout(),
task.getTaskCancellationTimeout());
+ assertThat(task.getTaskCancellationInterval())
+ .isEqualTo(executionConfig.getTaskCancellationInterval());
+ assertThat(task.getTaskCancellationTimeout())
+ .isEqualTo(executionConfig.getTaskCancellationTimeout());
task.getExecutingThread().interrupt();
task.getExecutingThread().join();
}
@Test
- public void testTerminationFutureCompletesOnNormalExecution() throws
Exception {
+ void testTerminationFutureCompletesOnNormalExecution() throws Exception {
final Task task =
createTaskBuilder()
.setInvokable(InvokableBlockingWithTrigger.class)
@@ -1159,17 +1146,17 @@ public class TaskTest extends TestLogger {
// wait till the task is in invoke
awaitInvokableLatch(task);
- assertFalse(task.getTerminationFuture().isDone());
+ assertThat(task.getTerminationFuture()).isNotDone();
triggerInvokableLatch(task);
task.getExecutingThread().join();
- assertEquals(ExecutionState.FINISHED,
task.getTerminationFuture().getNow(null));
+
assertThat(task.getTerminationFuture().getNow(null)).isSameAs(ExecutionState.FINISHED);
}
@Test
- public void testTerminationFutureCompletesOnImmediateCancellation() throws
Exception {
+ void testTerminationFutureCompletesOnImmediateCancellation() throws
Exception {
final Task task =
createTaskBuilder()
.setInvokable(InvokableBlockingInInvoke.class)
@@ -1178,18 +1165,18 @@ public class TaskTest extends TestLogger {
task.cancelExecution();
- assertFalse(task.getTerminationFuture().isDone());
+ assertThat(task.getTerminationFuture()).isNotDone();
// run the task asynchronous
task.startTaskThread();
task.getExecutingThread().join();
- assertEquals(ExecutionState.CANCELED,
task.getTerminationFuture().getNow(null));
+
assertThat(task.getTerminationFuture().getNow(null)).isSameAs(ExecutionState.CANCELED);
}
@Test
- public void testTerminationFutureCompletesOnErrorInInvoke() throws
Exception {
+ void testTerminationFutureCompletesOnErrorInInvoke() throws Exception {
final Task task =
createTaskBuilder()
.setInvokable(InvokableWithExceptionInInvoke.class)
@@ -1201,17 +1188,17 @@ public class TaskTest extends TestLogger {
task.getExecutingThread().join();
- assertEquals(ExecutionState.FAILED,
task.getTerminationFuture().getNow(null));
+
assertThat(task.getTerminationFuture().getNow(null)).isSameAs(ExecutionState.FAILED);
}
@Test
- public void testNoBackPressureIfTaskNotStarted() throws Exception {
+ void testNoBackPressureIfTaskNotStarted() throws Exception {
final Task task =
createTaskBuilder().build(Executors.directExecutor());
- assertFalse(task.isBackPressured());
+ assertThat(task.isBackPressured()).isFalse();
}
@Test
- public void testDeclineCheckpoint() throws Exception {
+ void testDeclineCheckpoint() throws Exception {
TestCheckpointResponder testCheckpointResponder = new
TestCheckpointResponder();
final Task task =
createTaskBuilder()
@@ -1227,7 +1214,7 @@ public class TaskTest extends TestLogger {
task.startTaskThread();
try {
awaitInvokableLatch(task);
- assertEquals(ExecutionState.RUNNING, task.getExecutionState());
+
assertThat(task.getExecutionState()).isSameAs(ExecutionState.RUNNING);
assertCheckpointDeclined(
task,
@@ -1248,7 +1235,7 @@ public class TaskTest extends TestLogger {
triggerInvokableLatch(task);
task.getExecutingThread().join();
}
- assertEquals(ExecutionState.FINISHED,
task.getTerminationFuture().getNow(null));
+
assertThat(task.getTerminationFuture().getNow(null)).isSameAs(ExecutionState.FINISHED);
}
private void testChannelStateWriterCloses(Class<? extends
TriggerLatchInvokable> invokable)
@@ -1263,19 +1250,19 @@ public class TaskTest extends TestLogger {
awaitInvokableLatch(task);
ChannelStateWriterWithCloseTracker channelStateWriter =
(ChannelStateWriterWithCloseTracker)
task.getChannelStateWriter();
- assertFalse(channelStateWriter.isClosed());
+ assertThat(channelStateWriter.isClosed()).isFalse();
triggerInvokableLatch(task);
task.getExecutingThread().join();
- assertTrue(channelStateWriter.isClosed());
+ assertThat(channelStateWriter.isClosed()).isTrue();
}
@Test
- public void testChannelStateWriterClosesOnSuccess() throws Exception {
+ void testChannelStateWriterClosesOnSuccess() throws Exception {
testChannelStateWriterCloses(ChannelStateWriterSetterInvokable.class);
}
@Test
- public void testChannelStateWriterClosesOnFailure() throws Exception {
+ void testChannelStateWriterClosesOnFailure() throws Exception {
testChannelStateWriterCloses(FailingChannelStateWriterSetterInvokable.class);
}
@@ -1289,16 +1276,16 @@ public class TaskTest extends TestLogger {
CheckpointType.CHECKPOINT,
CheckpointStorageLocationReference.getDefault());
task.triggerCheckpointBarrier(checkpointId, 1, checkpointOptions);
- assertEquals(1, testCheckpointResponder.getDeclineReports().size());
- assertEquals(
- checkpointId,
testCheckpointResponder.getDeclineReports().get(0).getCheckpointId());
- assertEquals(
- failureReason,
- testCheckpointResponder
- .getDeclineReports()
- .get(0)
- .getCause()
- .getCheckpointFailureReason());
+ assertThat(testCheckpointResponder.getDeclineReports()).hasSize(1);
+
assertThat(testCheckpointResponder.getDeclineReports().get(0).getCheckpointId())
+ .isEqualTo(checkpointId);
+ assertThat(
+ testCheckpointResponder
+ .getDeclineReports()
+ .get(0)
+ .getCause()
+ .getCheckpointFailureReason())
+ .isEqualTo(failureReason);
testCheckpointResponder.clear();
}
@@ -1339,7 +1326,7 @@ public class TaskTest extends TestLogger {
@Override
public void updateTaskExecutionState(TaskExecutionState
taskExecutionState) {
- assertTrue(queue.offer(taskExecutionState));
+ assertThat(queue.offer(taskExecutionState)).isTrue();
}
private void validateListenerMessage(ExecutionState state, Task task,
Throwable error) {
@@ -1347,16 +1334,16 @@ public class TaskTest extends TestLogger {
// we may have to wait for a bit to give the actors time to
receive the message
// and put it into the queue
final TaskExecutionState taskState = queue.take();
- assertNotNull("There is no additional listener message",
state);
+ assertThat(state).as("There is no additional listener
message").isNotNull();
- assertEquals(task.getExecutionId(), taskState.getID());
- assertEquals(state, taskState.getExecutionState());
+ assertThat(taskState.getID()).isEqualTo(task.getExecutionId());
+ assertThat(taskState.getExecutionState()).isSameAs(state);
final Throwable t =
taskState.getError(getClass().getClassLoader());
if (error == null) {
- assertNull(t);
+ assertThat(t).isNull();
} else {
- assertEquals(error.toString(), t.toString());
+ assertThat(t).hasToString(error.toString());
}
} catch (InterruptedException e) {
fail("interrupted");
