ableegoldman commented on a change in pull request #10609:
URL: https://github.com/apache/kafka/pull/10609#discussion_r646986783
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java
##########
@@ -593,6 +575,111 @@ public void shouldLogTempDirMessage() {
}
}
+ /************* Named Topology Tests *************/
+
+ @Test
+ public void shouldCreateTaskDirectoriesUnderNamedTopologyDirs() throws
IOException {
+ initializeStateDirectory(true, true);
+
+ directory.getOrCreateDirectoryForTask(new TaskId(0, 0, "topology1"));
+ directory.getOrCreateDirectoryForTask(new TaskId(0, 1, "topology1"));
+ directory.getOrCreateDirectoryForTask(new TaskId(0, 0, "topology2"));
+
+ assertThat(new File(appDir, "__topology1__").exists(), is(true));
+ assertThat(new File(appDir, "__topology1__").isDirectory(), is(true));
+ assertThat(new File(appDir, "__topology2__").exists(), is(true));
+ assertThat(new File(appDir, "__topology2__").isDirectory(), is(true));
+
+ assertThat(new File(new File(appDir, "__topology1__"),
"0_0").exists(), is(true));
+ assertThat(new File(new File(appDir, "__topology1__"),
"0_0").isDirectory(), is(true));
+ assertThat(new File(new File(appDir, "__topology1__"),
"0_1").exists(), is(true));
+ assertThat(new File(new File(appDir, "__topology1__"),
"0_1").isDirectory(), is(true));
+ assertThat(new File(new File(appDir, "__topology2__"),
"0_0").exists(), is(true));
+ assertThat(new File(new File(appDir, "__topology2__"),
"0_0").isDirectory(), is(true));
+ }
+
+ @Test
+ public void shouldOnlyListNonEmptyTaskDirectoriesInNamedTopologies()
throws IOException {
+ initializeStateDirectory(true, true);
+
+ TestUtils.tempDirectory(appDir.toPath(), "foo");
+ final TaskDirectory taskDir1 = new
TaskDirectory(directory.getOrCreateDirectoryForTask(new TaskId(0, 0,
"topology1")), "topology1");
+ final TaskDirectory taskDir2 = new
TaskDirectory(directory.getOrCreateDirectoryForTask(new TaskId(0, 1,
"topology1")), "topology1");
+ final TaskDirectory taskDir3 = new
TaskDirectory(directory.getOrCreateDirectoryForTask(new TaskId(0, 0,
"topology2")), "topology2");
+
+ final File storeDir = new File(taskDir1.file(), "store");
+ assertTrue(storeDir.mkdir());
+
+ assertThat(new HashSet<>(directory.listAllTaskDirectories()),
equalTo(mkSet(taskDir1, taskDir2, taskDir3)));
+ assertThat(directory.listNonEmptyTaskDirectories(),
equalTo(singletonList(taskDir1)));
+
+ Utils.delete(taskDir1.file());
+
+ assertThat(new HashSet<>(directory.listAllTaskDirectories()),
equalTo(mkSet(taskDir2, taskDir3)));
+ assertThat(directory.listNonEmptyTaskDirectories(),
equalTo(emptyList()));
+ }
+
+ @Test
+ public void shouldRemoveNonEmptyNamedTopologyDirsWhenCallingClean() throws
Exception {
+ initializeStateDirectory(true, true);
+ final File taskDir = directory.getOrCreateDirectoryForTask(new
TaskId(2, 0, "topology1"));
+ final File namedTopologyDir = new File(appDir, "__topology1__");
+
+ assertThat(taskDir.exists(), is(true));
+ assertThat(namedTopologyDir.exists(), is(true));
+ directory.clean();
+ assertThat(taskDir.exists(), is(false));
+ assertThat(namedTopologyDir.exists(), is(false));
+ }
+
+ @Test
+ public void shouldRemoveEmptyNamedTopologyDirsWhenCallingClean() throws
IOException {
+ initializeStateDirectory(true, true);
+ final File namedTopologyDir = new File(appDir, "__topology1__");
+ assertThat(namedTopologyDir.mkdir(), is(true));
+ assertThat(namedTopologyDir.exists(), is(true));
+ directory.clean();
+ assertThat(namedTopologyDir.exists(), is(false));
+ }
+
+ @Test
+ public void
shouldNotRemoveDirsThatDoNotMatchNamedTopologyDirsWhenCallingClean() throws
IOException {
+ initializeStateDirectory(true, true);
+ final File someDir = new File(appDir,
"_not-a-valid-named-topology_dir_name_");
+ assertThat(someDir.mkdir(), is(true));
+ assertThat(someDir.exists(), is(true));
+ directory.clean();
+ assertThat(someDir.exists(), is(true));
+ }
+
+ @Test
+ public void
shouldCleanupObsoleteTaskDirectoriesInNamedTopologiesAndDeleteTheParentDirectories()
throws IOException {
Review comment:
Ack, will add this test in the next PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
