Skip to content

[FLINK-40561][changelog] Migrate flink-statebackend-changelog tests to JUnit 5 and AssertJ #29113

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Alwaysgaurav1 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-40561][changelog] Migrate flink-statebackend-changelog tests to JUnit 5 and AssertJ #29113

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 38 additions & 33 deletions ...-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogDelegateStateTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,74 +32,76 @@
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackend;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.util.IOUtils;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

import java.util.Arrays;
import java.util.List;
import java.util.function.Supplier;

import static org.apache.flink.state.changelog.ChangelogStateBackendTestUtils.DummyCheckpointingStorageAccess;
import static org.apache.flink.state.changelog.ChangelogStateBackendTestUtils.createKeyedBackend;
import static org.junit.Assert.assertSame;
import static org.assertj.core.api.Assertions.assertThat;

/** Tests for {@link ChangelogStateBackend} delegating state accesses. */
@RunWith(Parameterized.class)
public class ChangelogDelegateStateTest {
@ExtendWith(ParameterizedTestExtension.class)
class ChangelogDelegateStateTest {
private MockEnvironment env;

@Parameterized.Parameters
@Parameters
public static List<Supplier<AbstractStateBackend>> delegatedStateBackend() {
return Arrays.asList(HashMapStateBackend::new, EmbeddedRocksDBStateBackend::new);
}

@Parameterized.Parameter public Supplier<AbstractStateBackend> backend;
@Parameter public Supplier<AbstractStateBackend> backend;

@Before
public void before() {
@BeforeEach
void before() {
env = MockEnvironment.builder().build();
env.setCheckpointStorageAccess(new DummyCheckpointingStorageAccess());
}

@After
public void after() {
@AfterEach
void after() {
IOUtils.closeQuietly(env);
}

@Test
public void testDelegatingValueState() throws Exception {
@TestTemplate
void testDelegatingValueState() throws Exception {
testDelegatingState(
new ValueStateDescriptor<>("id", String.class), ChangelogValueState.class);
}

@Test
public void testDelegatingListState() throws Exception {
@TestTemplate
void testDelegatingListState() throws Exception {
testDelegatingState(
new ListStateDescriptor<>("id", String.class), ChangelogListState.class);
}

@Test
public void testDelegatingMapState() throws Exception {
@TestTemplate
void testDelegatingMapState() throws Exception {
testDelegatingState(
new MapStateDescriptor<>("id", Integer.class, String.class),
ChangelogMapState.class);
}

@Test
public void testDelegatingReducingState() throws Exception {
@TestTemplate
void testDelegatingReducingState() throws Exception {
testDelegatingState(
new ReducingStateDescriptor<>(
"id", (value1, value2) -> value1 + "," + value2, String.class),
ChangelogReducingState.class);
}

@Test
public void testDelegatingAggregatingState() throws Exception {
@TestTemplate
void testDelegatingAggregatingState() throws Exception {
testDelegatingState(
new AggregatingStateDescriptor<>(
"my-state",
Expand All @@ -121,15 +123,18 @@ private void testDelegatingState(StateDescriptor descriptor, Class<?> stateClass
changelogBackend.getPartitionedState(
VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, descriptor);

assertSame(state.getClass(), stateClass);
assertSame(
((AbstractChangelogState<?, ?, ?, ?>) state).getDelegatedState().getClass(),
delegatedBackend
.getPartitionedState(
VoidNamespace.INSTANCE,
VoidNamespaceSerializer.INSTANCE,
descriptor)
.getClass());
assertThat(state.getClass()).isSameAs(stateClass);
assertThat(
((AbstractChangelogState<?, ?, ?, ?>) state)
.getDelegatedState()
.getClass())
.isSameAs(
delegatedBackend
.getPartitionedState(
VoidNamespace.INSTANCE,
VoidNamespaceSerializer.INSTANCE,
descriptor)
.getClass());
} finally {
if (delegatedBackend != null) {
delegatedBackend.dispose();
Expand Down
40 changes: 20 additions & 20 deletions ...ngelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,25 +41,25 @@
import org.apache.flink.state.changelog.ChangelogStateBackendTestUtils.DummyCheckpointingStorageAccess;
import org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationRunnable;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameter;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;

import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

import java.io.IOException;
import java.util.Optional;
import java.util.concurrent.RunnableFuture;

import static java.util.Collections.emptyList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;

/** {@link ChangelogKeyedStateBackend} test. */
@RunWith(Parameterized.class)
public class ChangelogKeyedStateBackendTest {
@ExtendWith(ParameterizedTestExtension.class)
class ChangelogKeyedStateBackendTest {

@Parameterized.Parameters(name = "checkpointID={0}, materializationId={1}")
@Parameters(name = "checkpointID={0}, materializationId={1}")
public static Object[][] parameters() {
return new Object[][] {
{0L, 200L},
Expand All @@ -73,8 +73,8 @@ public static Object[][] parameters() {
@Parameter(1)
public long materializationId;

@Test
public void testCheckpointConfirmation() throws Exception {
@TestTemplate
void testCheckpointConfirmation() throws Exception {
MockKeyedStateBackend<Integer> mock = createMock();
ChangelogKeyedStateBackend<Integer> changelog = createChangelog(mock);
try {
Expand All @@ -83,16 +83,16 @@ public void testCheckpointConfirmation() throws Exception {
checkpoint(changelog, checkpointId).get().discardState();

changelog.notifyCheckpointComplete(checkpointId);
assertEquals(materializationId, mock.getLastCompletedCheckpointID());
assertThat(mock.getLastCompletedCheckpointID()).isEqualTo(materializationId);

} finally {
changelog.close();
changelog.dispose();
}
}

@Test
public void testInitMaterialization() throws Exception {
@TestTemplate
void testInitMaterialization() throws Exception {
MockKeyedStateBackend<Integer> delegatedBackend = createMock();
ChangelogKeyedStateBackend<Integer> backend = createChangelog(delegatedBackend);

Expand All @@ -103,25 +103,25 @@ public void testInitMaterialization() throws Exception {

runnable = backend.initMaterialization();
// 1. should trigger first materialization
assertTrue("first materialization should be trigger.", runnable.isPresent());
assertThat(runnable).as("first materialization should be trigger.").isPresent();

appendMockStateChange(backend); // ensure there is non-materialized changelog

// 2. should not trigger new one until the previous one has been confirmed or failed
assertFalse(backend.initMaterialization().isPresent());
assertThat(backend.initMaterialization()).isNotPresent();

backend.handleMaterializationFailureOrCancellation(
runnable.get().getMaterializationID(),
runnable.get().getMaterializedTo(),
null);
runnable = backend.initMaterialization();
// 3. should trigger new one after previous one failed
assertTrue(runnable.isPresent());
assertThat(runnable).isPresent();

appendMockStateChange(backend); // ensure there is non-materialized changelog

// 4. should not trigger new one until the previous one has been confirmed or failed
assertFalse(backend.initMaterialization().isPresent());
assertThat(backend.initMaterialization()).isNotPresent();

backend.handleMaterializationResult(
SnapshotResult.empty(),
Expand All @@ -130,7 +130,7 @@ public void testInitMaterialization() throws Exception {
checkpoint(backend, checkpointId).get().discardState();
backend.notifyCheckpointComplete(checkpointId);
// 5. should trigger new one after previous one has been confirmed
assertTrue(backend.initMaterialization().isPresent());
assertThat(backend.initMaterialization()).isPresent();
} finally {
backend.close();
backend.dispose();
Expand Down
36 changes: 17 additions & 19 deletions ...kend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogListStateTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
import org.apache.flink.util.function.FunctionWithException;
import org.apache.flink.util.function.ThrowingConsumer;

import org.junit.Test;
import org.junit.jupiter.api.Test;

import java.util.ArrayList;
import java.util.Arrays;
Expand All @@ -38,50 +38,48 @@

import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;

/** ChangelogListState Test. */
@SuppressWarnings({"rawtypes", "unchecked"})
public class ChangelogListStateTest {
class ChangelogListStateTest {

@Test
public void testValuesIterator() throws Exception {
void testValuesIterator() throws Exception {
testIterator(singletonList("value"), state -> state.get().iterator(), "value");
}

@Test
public void testPutRecorded() throws Exception {
void testPutRecorded() throws Exception {
testRecorded(
emptyList(),
state -> state.add("x"),
logger -> assertTrue(logger.stateElementAdded));
logger -> assertThat(logger.stateElementAdded).isTrue());
}

@Test
public void testAddAllRecorded() throws Exception {
void testAddAllRecorded() throws Exception {
List<String> list = Arrays.asList("a", "b", "c");
testRecorded(
emptyList(),
state -> state.addAll(list),
logger -> assertEquals(list, logger.state));
logger -> assertThat(logger.state).isEqualTo(list));
}

@Test
public void testGetNotRecorded() throws Exception {
void testGetNotRecorded() throws Exception {
testRecorded(
singletonList("x"),
ChangelogListState::get,
logger -> assertFalse(logger.anythingChanged()));
logger -> assertThat(logger.anythingChanged()).isFalse());
}

@Test
public void testClearRecorded() throws Exception {
void testClearRecorded() throws Exception {
testRecorded(
singletonList("x"),
ChangelogListState::clear,
logger -> assertTrue(logger.stateCleared));
logger -> assertThat(logger.stateCleared).isTrue());
}

private <T> void testIterator(
Expand All @@ -94,15 +92,15 @@ private <T> void testIterator(

Iterator iterator = iteratorSupplier.apply(state);
for (T el : elements) {
assertTrue(iterator.hasNext());
assertEquals(el, iterator.next());
assertThat(iterator.hasNext()).isTrue();
assertThat(iterator.next()).isEqualTo(el);
iterator.remove();
}

assertFalse(iterator.hasNext());
assertTrue(state.getInternal().isEmpty());
assertThat(iterator.hasNext()).isFalse();
assertThat(state.getInternal().isEmpty()).isTrue();
// changes to the rocksdb list iterator are not propagated back - expect the same here
assertFalse(logger.stateElementRemoved);
assertThat(logger.stateElementRemoved).isFalse();
}

private void testRecorded(
Expand Down
Loading