class PutHiveStreaming extends AbstractProcessor {

    @OnStopped
    public void cleanup() {
        ComponentLog log = getLogger();
        sendHeartBeat.set(false);
        for (Map.Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
            try {
                HiveWriter w = entry.getValue();
                w.flushAndClose();
            } catch (Exception ex) {
                log.warn("Error while closing writer to " + entry.getKey() + ". Exception follows.", ex);
                if (ex instanceof InterruptedException) {
                    Thread.currentThread().interrupt();
                }
            }
        }

        callTimeoutPool.shutdown();
        try {
            while (!callTimeoutPool.isTerminated()) {
                callTimeoutPool.awaitTermination(options.getCallTimeOut(), TimeUnit.MILLISECONDS);
            }
        } catch (Throwable t) {
            log.warn("shutdown interrupted on " + callTimeoutPool, t);
        }

        callTimeoutPool = null;
        ugi = null;
        hiveConfigurator.stopRenewer();
    }

}