Skip to content

MINOR: cleanup and add tests to StateDirectoryTest #8304

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

Merged
guozhangwang merged 3 commits into from
Mar 17, 2020
Merged

MINOR: cleanup and add tests to StateDirectoryTest #8304

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7d1585b
minor cleanup and add tests
ableegoldman Mar 16, 2020
6460f72
minor cleanup
ableegoldman Mar 16, 2020
76e12b2
remove unused import
ableegoldman Mar 17, 2020
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
31 changes: 21 additions & 10 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand All @@ -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;
}
Expand Down
98 changes: 58 additions & 40 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -58,7 +59,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();
Expand All @@ -76,12 +77,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);
}

Expand All @@ -102,7 +103,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);

Expand All @@ -113,17 +114,14 @@ public void shouldLockTaskStateDirectory() throws Exception {
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);
}
}

@Test
public void shouldBeTrueIfAlreadyHoldsLock() throws Exception {
public void shouldBeTrueIfAlreadyHoldsLock() throws IOException {
final TaskId taskId = new TaskId(0, 0);
directory.directoryForTask(taskId);
directory.lock(taskId);
Expand All @@ -135,13 +133,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
Expand Down Expand Up @@ -175,29 +173,24 @@ 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);

try {
directory.directoryForTask(taskId);
fail("Should have thrown ProcessorStateException");
} catch (final ProcessorStateException expected) {
// swallow
}
assertThrows(ProcessorStateException.class, () -> directory.directoryForTask(taskId));
}

@Test
public void shouldNotLockDeletedDirectory() throws Exception {
public void shouldNotLockDeletedDirectory() throws IOException {
final TaskId taskId = new TaskId(0, 0);

Utils.delete(stateDir);
assertFalse(directory.lock(taskId));
}

@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);
Expand All @@ -216,19 +209,16 @@ public void shouldLockMultipleTaskDirectories() throws Exception {
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);
}
}

@Test
public void shouldReleaseTaskStateDirectoryLock() throws Exception {
public void shouldReleaseTaskStateDirectoryLock() throws IOException {
final TaskId taskId = new TaskId(0, 0);
final File taskDirectory = directory.directoryForTask(taskId);

Expand All @@ -246,7 +236,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);
Expand Down Expand Up @@ -306,6 +296,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");
Expand Down Expand Up @@ -343,26 +365,22 @@ public void shouldCreateDirectoriesIfParentDoesntExist() {
}

@Test
public void shouldLockGlobalStateDirectory() throws Exception {
directory.lockGlobalState();

public void shouldLockGlobalStateDirectory() throws IOException {
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();
}
}

@Test
public void shouldUnlockGlobalStateDirectory() throws Exception {
public void shouldUnlockGlobalStateDirectory() throws IOException {
directory.lockGlobalState();
directory.unlockGlobalState();

Expand Down Expand Up @@ -439,36 +457,36 @@ 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);
assertFalse(taskDirectory.exists());
}

@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());
}
Expand Down