Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix leaking Status Path nodes in ZK 

- remove  ZK status path nodes for workers after they are removed
This commit is contained in:
Nishant 2015-07-09 17:20:09 +05:30
parent 66d105940d
commit 8d7a566bae
2 changed files with 16 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
indexing-service/src/main/java/io/druid/indexing/overlord/RemoteTaskRunner.java
View File

@ -878,17 +878,12 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogStreamer
for (String assignedTask : tasksToFail) { for (String assignedTask : tasksToFail) {
String taskPath = JOINER.join(indexerZkConfig.getTasksPath(), worker, assignedTask); String taskPath = JOINER.join(indexerZkConfig.getTasksPath(), worker, assignedTask);
String statusPath = JOINER.join(indexerZkConfig.getStatusPath(), worker, assignedTask); String statusPath = JOINER.join(indexerZkConfig.getStatusPath(), worker, assignedTask);
try { if (cf.checkExists().forPath(taskPath) != null) {
if (cf.checkExists().forPath(taskPath) != null) { cf.delete().guaranteed().forPath(taskPath);
cf.delete().guaranteed().forPath(taskPath);
}
if (cf.checkExists().forPath(statusPath) != null) {
cf.delete().guaranteed().forPath(statusPath);
}
} }
catch (Exception e) {
throw Throwables.propagate(e); if (cf.checkExists().forPath(statusPath) != null) {
cf.delete().guaranteed().forPath(statusPath);
} }
log.info("Failing task[%s]", assignedTask); log.info("Failing task[%s]", assignedTask);
@ -899,8 +894,17 @@ public class RemoteTaskRunner implements TaskRunner, TaskLogStreamer
log.warn("RemoteTaskRunner has no knowledge of task[%s]", assignedTask); log.warn("RemoteTaskRunner has no knowledge of task[%s]", assignedTask);
} }
} }
// worker is gone, remove worker task status announcements path.
String workerStatusPath = JOINER.join(indexerZkConfig.getStatusPath(), worker);
if (cf.checkExists().forPath(workerStatusPath) != null) {
cf.delete().guaranteed().forPath(JOINER.join(indexerZkConfig.getStatusPath(), worker));
}
} }
finally { catch (Exception e) {
log.makeAlert("Exception while cleaning up worker[%s]", worker).emit();
throw Throwables.propagate(e);
} finally {
removedWorkerCleanups.remove(worker); removedWorkerCleanups.remove(worker);
} }
} }

1
indexing-service/src/test/java/io/druid/indexing/overlord/RemoteTaskRunnerTest.java
View File

@ -379,6 +379,7 @@ public class RemoteTaskRunnerTest
} }
) )
); );
Assert.assertNull(cf.checkExists().forPath(statusPath));
} }
@Test @Test