be nice and close all resources on shutdown

This commit is contained in:
Christian Meier 2015-11-02 20:20:43 +01:00
parent 439e8f635e
commit 98af950498
3 changed files with 23 additions and 2 deletions

View File

@ -79,6 +79,7 @@ class KafkaPoller extends Thread {
Thread.sleep(delay.value()) Thread.sleep(delay.value())
} }
disconnectConsumers()
} }
private void slower(Delay delay) { private void slower(Delay delay) {
@ -178,7 +179,6 @@ class KafkaPoller extends Thread {
*/ */
void die() { void die() {
this.keepRunning = false this.keepRunning = false
disconnectConsumers()
} }
private void disconnectConsumers() { private void disconnectConsumers() {

View File

@ -126,8 +126,9 @@ class Main {
/* Assuming that most people aren't needing to run Storm-based watchers /* Assuming that most people aren't needing to run Storm-based watchers
* as well * as well
*/ */
KafkaSpoutTreeWatcher stormWatcher = null
if (cli.hasOption('s')) { if (cli.hasOption('s')) {
KafkaSpoutTreeWatcher stormWatcher = new KafkaSpoutTreeWatcher(client, stormWatcher = new KafkaSpoutTreeWatcher(client,
watchedTopics, watchedTopics,
consumerOffsets) consumerOffsets)
stormWatcher.onConsumerData << gaugeRegistrar stormWatcher.onConsumerData << gaugeRegistrar
@ -155,6 +156,18 @@ class Main {
/* Start the reporter if we've got it */ /* Start the reporter if we've got it */
reporter?.start(delayInSeconds, TimeUnit.SECONDS) reporter?.start(delayInSeconds, TimeUnit.SECONDS)
// shutdown threads
Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
Main.logger.info("showdown threads")
poller.die()
consumerWatcher.close()
if (stormWatcher != null) {
stormWatcher.close()
}
poller.join()
}
});
logger.info('Starting wait loop...') logger.info('Starting wait loop...')
synchronized(this) { synchronized(this) {
wait() wait()

View File

@ -47,5 +47,13 @@ abstract class AbstractTreeWatcher implements TreeCacheListener {
return this return this
} }
/**
* Close our internal cache and return ourselves for API cleanliness
*/
AbstractTreeWatcher close() {
this.cache?.close()
return this
}
abstract void childEvent(CuratorFramework client, TreeCacheEvent event) abstract void childEvent(CuratorFramework client, TreeCacheEvent event)
} }