Improve test coverage GlobalStateManagerImpl
Add coverage for exception paths in `initialize()`
Author: Damian Guy
Reviewers: Eno Thereska, Matthias J. Sax, Guozhang Wang
Closes from dguy/kafka-4649
@@ -68,6 +68,7 @@ public class GlobalStateManagerImplTest {
private MockConsumer<byte[], byte[]> consumer;
private File checkpointFile;
private final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
+ private ProcessorTopology topology;
@Before
public void before() throws IOException {
@@ -80,7 +81,7 @@ public class GlobalStateManagerImplTest {
storeToProcessorNode.put(store1, new MockProcessorNode(-1));
store2 = new NoOpReadOnlyStore("t2-store");
storeToProcessorNode.put(store2, new MockProcessorNode(-1));
- final ProcessorTopology topology = new ProcessorTopology(Collections.<ProcessorNode>emptyList(),
+ topology = new ProcessorTopology(Collections.<ProcessorNode>emptyList(),
Collections.<String, SourceNode>emptyMap(),
Collections.<String, SinkNode>emptyMap(),
Collections.<StateStore>emptyList(),
@@ -107,7 +108,7 @@ public class GlobalStateManagerImplTest {
}
@Test(expected = LockException.class)
- public void shouldThrowStreamsExceptionIfCantGetLock() throws Exception {
+ public void shouldThrowLockExceptionIfCantGetLock() throws Exception {
final StateDirectory stateDir = new StateDirectory("appId", stateDirPath);
try {
stateDir.lockGlobalState(1);
@@ -135,10 +136,7 @@ public class GlobalStateManagerImplTest {
@Test(expected = StreamsException.class)
public void shouldThrowStreamsExceptionIfFailedToReadCheckpointedOffsets() throws Exception {
- final File checkpointFile = new File(stateManager.baseDir(), ProcessorStateManager.CHECKPOINT_FILE_NAME);
- try (final FileOutputStream stream = new FileOutputStream(checkpointFile)) {
- stream.write("0\n1\nblah".getBytes());
- }
+ writeCorruptCheckpoint();
stateManager.initialize(context);
}
@@ -361,10 +359,7 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldReleaseLockIfExceptionWhenLoadingCheckpoints() throws Exception {
- final File checkpointFile = new File(stateManager.baseDir(), ProcessorStateManager.CHECKPOINT_FILE_NAME);
- try (final FileOutputStream stream = new FileOutputStream(checkpointFile)) {
- stream.write("0\n1\nblah".getBytes());
- }
+ writeCorruptCheckpoint();
try {
stateManager.initialize(context);
} catch (StreamsException e) {
@@ -379,6 +374,26 @@ public class GlobalStateManagerImplTest {
}
}
+
+ @Test(expected = LockException.class)
+ public void shouldThrowLockExceptionIfIOExceptionCaughtWhenTryingToLockStateDir() throws Exception {
+ stateManager = new GlobalStateManagerImpl(topology, consumer, new StateDirectory("appId", stateDirPath) {
+ @Override
+ public boolean lockGlobalState(final int retry) throws IOException {
+ throw new IOException("KABOOM!");
+ }
+ });
+
+ stateManager.initialize(context);
+ }
+
+ private void writeCorruptCheckpoint() throws IOException {
+ final File checkpointFile = new File(stateManager.baseDir(), ProcessorStateManager.CHECKPOINT_FILE_NAME);
+ try (final FileOutputStream stream = new FileOutputStream(checkpointFile)) {
+ stream.write("0\n1\nfoo".getBytes());
+ }
+ }
+
private void initializeConsumer(final long numRecords, final long startOffset, final TopicPartition topicPartition) {
final HashMap<TopicPartition, Long> startOffsets = new HashMap<>();
startOffsets.put(topicPartition, 1L);
