class PutHiveStreaming extends AbstractSessionFactoryProcessor {

    @OnStopped
    public void cleanup() {
        validationResourceHolder.set(null); // trigger re-validation of resources

        ComponentLog log = getLogger();
        sendHeartBeat.set(false);
        for(Map<HiveEndPoint, HiveWriter> allWriters : threadWriterList) {
            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();
                    }
                }
            }
            allWriters.clear();
        }

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

        ugi = null;
    }

}