From 7d1585b8e414e822f715732726eb903c43fb5c5d Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Mon, 16 Mar 2020 14:32:20 -0700 Subject: [PATCH 1/3] minor cleanup and add tests --- .../processor/internals/StateDirectory.java | 31 +++++--- .../internals/StateDirectoryTest.java | 70 ++++++++++++++----- 2 files changed, 72 insertions(+), 29 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java index 41c7c4b6f478c..0a1e715ccdc35 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java @@ -361,14 +361,19 @@ private synchronized void cleanRemovedTasks(final long cleanupDelayMs, * @return The list of all the non-empty local directories for stream tasks */ File[] listNonEmptyTaskDirectories() { - final File[] taskDirectories = !stateDir.exists() ? new File[0] : - stateDir.listFiles(pathname -> { - if (!pathname.isDirectory() || !PATH_NAME.matcher(pathname.getName()).matches()) { - return false; - } else { - return !taskDirEmpty(pathname); - } - }); + final File[] taskDirectories; + if (!hasPersistentStores || !stateDir.exists()) { + taskDirectories = new File[0]; + } else { + taskDirectories = + stateDir.listFiles(pathname -> { + if (!pathname.isDirectory() || !PATH_NAME.matcher(pathname.getName()).matches()) { + return false; + } else { + return !taskDirEmpty(pathname); + } + }); + } return taskDirectories == null ? new File[0] : taskDirectories; } @@ -378,8 +383,14 @@ File[] listNonEmptyTaskDirectories() { * @return The list of all the existing local directories for stream tasks */ File[] listAllTaskDirectories() { - final File[] taskDirectories = !stateDir.exists() ? new File[0] : - stateDir.listFiles(pathname -> pathname.isDirectory() && PATH_NAME.matcher(pathname.getName()).matches()); + final File[] taskDirectories; + if (!hasPersistentStores || !stateDir.exists()) { + taskDirectories = new File[0]; + } else { + taskDirectories = + stateDir.listFiles(pathname -> pathname.isDirectory() + && PATH_NAME.matcher(pathname.getName()).matches()); + } return taskDirectories == null ? new File[0] : taskDirectories; } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java index 827557a88faae..ce6d81180f408 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java @@ -58,7 +58,7 @@ public class StateDirectoryTest { private StateDirectory directory; private File appDir; - private void initializeStateDirectory(final boolean createStateDirectory) throws Exception { + private void initializeStateDirectory(final boolean createStateDirectory) throws IOException { stateDir = new File(TestUtils.IO_TMP_DIR, "kafka-" + TestUtils.randomString(5)); if (!createStateDirectory) { cleanup(); @@ -76,12 +76,12 @@ private void initializeStateDirectory(final boolean createStateDirectory) throws } @Before - public void before() throws Exception { + public void before() throws IOException { initializeStateDirectory(true); } @After - public void cleanup() throws Exception { + public void cleanup() throws IOException { Utils.delete(stateDir); } @@ -102,7 +102,7 @@ public void shouldCreateTaskStateDirectory() { } @Test - public void shouldLockTaskStateDirectory() throws Exception { + public void shouldLockTaskStateDirectory() throws IOException { final TaskId taskId = new TaskId(0, 0); final File taskDirectory = directory.directoryForTask(taskId); @@ -123,7 +123,7 @@ public void shouldLockTaskStateDirectory() throws Exception { } @Test - public void shouldBeTrueIfAlreadyHoldsLock() throws Exception { + public void shouldBeTrueIfAlreadyHoldsLock() throws IOException { final TaskId taskId = new TaskId(0, 0); directory.directoryForTask(taskId); directory.lock(taskId); @@ -135,13 +135,13 @@ public void shouldBeTrueIfAlreadyHoldsLock() throws Exception { } @Test - public void shouldBeAbleToUnlockEvenWithoutLocking() throws Exception { + public void shouldBeAbleToUnlockEvenWithoutLocking() throws IOException { final TaskId taskId = new TaskId(0, 0); directory.unlock(taskId); } @Test - public void shouldReportDirectoryEmpty() throws Exception { + public void shouldReportDirectoryEmpty() throws IOException { final TaskId taskId = new TaskId(0, 0); // when task dir first created, it should be empty @@ -175,7 +175,7 @@ public void shouldReportDirectoryEmpty() throws Exception { } @Test - public void shouldThrowProcessorStateException() throws Exception { + public void shouldThrowProcessorStateException() throws IOException { final TaskId taskId = new TaskId(0, 0); Utils.delete(stateDir); @@ -189,7 +189,7 @@ public void shouldThrowProcessorStateException() throws Exception { } @Test - public void shouldNotLockDeletedDirectory() throws Exception { + public void shouldNotLockDeletedDirectory() throws IOException { final TaskId taskId = new TaskId(0, 0); Utils.delete(stateDir); @@ -197,7 +197,7 @@ public void shouldNotLockDeletedDirectory() throws Exception { } @Test - public void shouldLockMultipleTaskDirectories() throws Exception { + public void shouldLockMultipleTaskDirectories() throws IOException { final TaskId taskId = new TaskId(0, 0); final File task1Dir = directory.directoryForTask(taskId); final TaskId taskId2 = new TaskId(1, 0); @@ -228,7 +228,7 @@ public void shouldLockMultipleTaskDirectories() throws Exception { } @Test - public void shouldReleaseTaskStateDirectoryLock() throws Exception { + public void shouldReleaseTaskStateDirectoryLock() throws IOException { final TaskId taskId = new TaskId(0, 0); final File taskDirectory = directory.directoryForTask(taskId); @@ -246,7 +246,7 @@ public void shouldReleaseTaskStateDirectoryLock() throws Exception { } @Test - public void shouldCleanUpTaskStateDirectoriesThatAreNotCurrentlyLocked() throws Exception { + public void shouldCleanUpTaskStateDirectoriesThatAreNotCurrentlyLocked() throws IOException { final TaskId task0 = new TaskId(0, 0); final TaskId task1 = new TaskId(1, 0); final TaskId task2 = new TaskId(2, 0); @@ -306,6 +306,38 @@ public void shouldNotRemoveNonTaskDirectoriesAndFiles() { assertTrue(otherDir.exists()); } + @Test + public void shouldReturnEmptyArrayForNonPersistentApp() throws IOException { + initializeStateDirectory(false); + assertTrue(Arrays.asList(directory.listAllTaskDirectories()).isEmpty()); + } + + @Test + public void shouldReturnEmptyArrayIfStateDirDoesntExist() throws IOException { + cleanup(); + assertFalse(stateDir.exists()); + assertTrue(Arrays.asList(directory.listAllTaskDirectories()).isEmpty()); + } + + @Test + public void shouldReturnEmptyArrayIfListFilesReturnsNull() { + stateDir = new File(TestUtils.IO_TMP_DIR, "kafka-" + TestUtils.randomString(5)); + directory = new StateDirectory( + new StreamsConfig(new Properties() { + { + put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId); + put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234"); + put(StreamsConfig.STATE_DIR_CONFIG, stateDir.getPath()); + } + }), + time, true); + appDir = new File(stateDir, applicationId); + + assertTrue(stateDir.renameTo(new File(TestUtils.IO_TMP_DIR, "state-renamed"))); + + assertTrue(Arrays.asList(directory.listAllTaskDirectories()).isEmpty()); + } + @Test public void shouldOnlyListNonEmptyTaskDirectories() { TestUtils.tempDirectory(stateDir.toPath(), "foo"); @@ -343,7 +375,7 @@ public void shouldCreateDirectoriesIfParentDoesntExist() { } @Test - public void shouldLockGlobalStateDirectory() throws Exception { + public void shouldLockGlobalStateDirectory() throws IOException { directory.lockGlobalState(); try ( @@ -362,7 +394,7 @@ public void shouldLockGlobalStateDirectory() throws Exception { } @Test - public void shouldUnlockGlobalStateDirectory() throws Exception { + public void shouldUnlockGlobalStateDirectory() throws IOException { directory.lockGlobalState(); directory.unlockGlobalState(); @@ -439,14 +471,14 @@ public void shouldCleanupAllTaskDirectoriesIncludingGlobalOne() { } @Test - public void shouldNotCreateBaseDirectory() throws Exception { + public void shouldNotCreateBaseDirectory() throws IOException { initializeStateDirectory(false); assertFalse(stateDir.exists()); assertFalse(appDir.exists()); } @Test - public void shouldNotCreateTaskStateDirectory() throws Exception { + public void shouldNotCreateTaskStateDirectory() throws IOException { initializeStateDirectory(false); final TaskId taskId = new TaskId(0, 0); final File taskDirectory = directory.directoryForTask(taskId); @@ -454,21 +486,21 @@ public void shouldNotCreateTaskStateDirectory() throws Exception { } @Test - public void shouldNotCreateGlobalStateDirectory() throws Exception { + public void shouldNotCreateGlobalStateDirectory() throws IOException { initializeStateDirectory(false); final File globalStateDir = directory.globalStateDir(); assertFalse(globalStateDir.exists()); } @Test - public void shouldLockTaskStateDirectoryWhenDirectoryCreationDisabled() throws Exception { + public void shouldLockTaskStateDirectoryWhenDirectoryCreationDisabled() throws IOException { initializeStateDirectory(false); final TaskId taskId = new TaskId(0, 0); assertTrue(directory.lock(taskId)); } @Test - public void shouldLockGlobalStateDirectoryWhenDirectoryCreationDisabled() throws Exception { + public void shouldLockGlobalStateDirectoryWhenDirectoryCreationDisabled() throws IOException { initializeStateDirectory(false); assertTrue(directory.lockGlobalState()); } From 6460f7213b2fb4c5c6c0e2c1d1ce02c8e281ecfa Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Mon, 16 Mar 2020 14:42:06 -0700 Subject: [PATCH 2/3] minor cleanup --- .../internals/StateDirectoryTest.java | 28 +++++-------------- 1 file changed, 7 insertions(+), 21 deletions(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java index ce6d81180f408..a971318b41e0e 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java @@ -47,6 +47,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -113,10 +114,7 @@ public void shouldLockTaskStateDirectory() throws IOException { new File(taskDirectory, StateDirectory.LOCK_FILE_NAME).toPath(), StandardOpenOption.CREATE, StandardOpenOption.WRITE) ) { - channel.tryLock(); - fail("shouldn't be able to lock already locked directory"); - } catch (final OverlappingFileLockException e) { - // swallow + assertThrows(OverlappingFileLockException.class, channel::tryLock); } finally { directory.unlock(taskId); } @@ -180,12 +178,7 @@ public void shouldThrowProcessorStateException() throws IOException { Utils.delete(stateDir); - try { - directory.directoryForTask(taskId); - fail("Should have thrown ProcessorStateException"); - } catch (final ProcessorStateException expected) { - // swallow - } + assertThrows(ProcessorStateException.class, () -> directory.directoryForTask(taskId)); } @Test @@ -216,11 +209,8 @@ public void shouldLockMultipleTaskDirectories() throws IOException { directory.lock(taskId); directory.lock(taskId2); - channel1.tryLock(); - channel2.tryLock(); - fail("shouldn't be able to lock already locked directory"); - } catch (final OverlappingFileLockException e) { - // swallow + assertThrows(OverlappingFileLockException.class, channel1::tryLock); + assertThrows(OverlappingFileLockException.class, channel2::tryLock); } finally { directory.unlock(taskId); directory.unlock(taskId2); @@ -376,18 +366,14 @@ public void shouldCreateDirectoriesIfParentDoesntExist() { @Test public void shouldLockGlobalStateDirectory() throws IOException { - directory.lockGlobalState(); - try ( final FileChannel channel = FileChannel.open( new File(directory.globalStateDir(), StateDirectory.LOCK_FILE_NAME).toPath(), StandardOpenOption.CREATE, StandardOpenOption.WRITE) ) { - channel.lock(); - fail("Should have thrown OverlappingFileLockException"); - } catch (final OverlappingFileLockException expcted) { - // swallow + directory.lockGlobalState(); + assertThrows(OverlappingFileLockException.class, channel::lock); } finally { directory.unlockGlobalState(); } From 76e12b299a3db98b2cdf2d1cd7a8fac264d69d33 Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Mon, 16 Mar 2020 17:32:26 -0700 Subject: [PATCH 3/3] remove unused import --- .../kafka/streams/processor/internals/StateDirectoryTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java index a971318b41e0e..b07d6e66f1313 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java @@ -49,7 +49,6 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; public class StateDirectoryTest {