private void processSnapshotTruncationRequestCreated(
            final WatchedEvent event) {
        loggingLog.info("Snapshot truncation leader received snapshot truncation request");
        // Get the truncation request ID which is the truncation request node path.
        final String truncReqId;
        try {
            List<String> children = m_zk.getChildren(event.getPath(), false);
            if (children.isEmpty()) {
                loggingLog.error("Unable to retrieve truncation snapshot request id from ZK, log can't be truncated");
                return;
            }
            truncReqId = ZKUtil.joinZKPath(event.getPath(), Collections.max(children));

        } catch (Exception e) {
            loggingLog.error("Unable to retrieve truncation snapshot request ID from ZK, log can't be truncated");
            return;
        }
        final long now = System.currentTimeMillis();
        final String nonce = Long.toString(now);
        //Allow nodes to check and see if the nonce incoming for a snapshot is
        //for a truncation snapshot. In that case they will mark the completion node
        //to be for a truncation snapshot. SnapshotCompletionMonitor notices the mark.
        // TRAIL [TruncSnap:7] write current ts to request zk node data
        try {
            ByteBuffer payload = ByteBuffer.allocate(8);
            payload.putLong(0, now);
            m_zk.setData(VoltZK.request_truncation_snapshot, payload.array(), -1);
        } catch (Exception e) {
            //Cause a cascading failure?
            VoltDB.crashLocalVoltDB("Setting data on the truncation snapshot request in ZK should never fail", true, e);
        }
        // for the snapshot save invocations
        JSONObject jsObj = new JSONObject();
        try {
            assert truncReqId != null;
            String sData = "";
            JSONObject jsData = new JSONObject();
            jsData.put("truncReqId", truncReqId);
            sData = jsData.toString();
            jsObj.put(SnapshotUtil.JSON_PATH, VoltDB.instance().getCommandLogSnapshotPath() );
            jsObj.put(SnapshotUtil.JSON_NONCE, nonce);
            jsObj.put(SnapshotUtil.JSON_PATH_TYPE, SnapshotPathType.SNAP_CL);
            jsObj.put("perPartitionTxnIds", retrievePerPartitionTransactionIds());
            jsObj.put("data", sData);
        } catch (JSONException e) {
            /*
             * Should never happen, so fail fast
             */
            VoltDB.crashLocalVoltDB("", true, e);
        }

        // for the snapshot save invocations
        long handle = m_nextCallbackHandle++;

        // for the snapshot save invocation
        m_procedureCallbacks.put(handle, new ProcedureCallback() {

            @Override
            public void clientCallback(ClientResponse clientResponse)
                    throws Exception {
                m_lastInitiationTs = null;
                if (clientResponse.getStatus() != ClientResponse.SUCCESS){
                    loggingLog.warn(
                            "Attempt to initiate a truncation snapshot was not successful: " +
                            clientResponse.getStatusString());
                    loggingLog.warn("Retrying log truncation snapshot in 5 minutes");
                    /*
                     * TRAIL [TruncSnap:8] (callback) on failed response try again in a few minute
                     */
                    m_es.schedule(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                processTruncationRequestEvent(event);
                            } catch (Exception e) {
                                VoltDB.crashLocalVoltDB("Error processing snapshot truncation request event", true, e);
                            }
                        }
                    }, 5, TimeUnit.MINUTES);
                    return;
                }

                final VoltTable results[] = clientResponse.getResults();
                final VoltTable result = results[0];
                boolean success = true;

                final String err = SnapshotUtil.didSnapshotRequestFailWithErr(results);
                if (err != null) {
                    if (err.trim().equalsIgnoreCase("SNAPSHOT IN PROGRESS")) {
                        loggingLog.info("Snapshot is in progress");
                    } else {
                        loggingLog.warn("Snapshot failed with failure response: " + err);
                    }
                    success = false;
                }

                //assert(result.getColumnName(1).equals("TABLE"));
                if (success) {
                    while (result.advanceRow()) {
                        if (!result.getString("RESULT").equals("SUCCESS")) {
                            success = false;
                            loggingLog.warn("Snapshot save feasibility test failed for host "
                                    + result.getLong("HOST_ID") + " table " + result.getString("TABLE") +
                                    " with error message " + result.getString("ERR_MSG"));
                        }
                    }
                }

                if (success) {
                    loggingLog.info("Snapshot initiation for log truncation was successful");

                    JSONObject obj = new JSONObject(clientResponse.getAppStatusString());
                    final long snapshotTxnId = Long.valueOf(obj.getLong("txnId"));
                    try {
                        boolean found = false;
                        ZKUtil.VoidCallback lastCallback = null;

                        for (String child: m_zk.getChildren(event.getPath(),false)) {
                            String requestId = ZKUtil.joinZKPath(event.getPath(), child);
                            found = found || requestId.equals(truncReqId);

                            lastCallback = new ZKUtil.VoidCallback();
                            m_zk.delete(requestId, -1, lastCallback, null);
                        }

                        if (lastCallback != null) {
                            try {
                                lastCallback.get();
                            } catch (KeeperException.NoNodeException ignoreIt) {
                            }
                        }
                        if (!found) {
                            VoltDB.crashLocalVoltDB(
                                    "Could not match truncations snapshot request id while atepting its removal", true, null);
                        }
                    } catch (Exception e) {
                        VoltDB.crashLocalVoltDB(
                                "Unexpected error deleting truncation snapshot request", true, e);
                    }

                    try {
                        TruncationSnapshotAttempt snapshotAttempt =
                            m_truncationSnapshotAttempts.get(snapshotTxnId);
                        if (snapshotAttempt == null) {
                            snapshotAttempt = new TruncationSnapshotAttempt();
                            m_truncationSnapshotAttempts.put(snapshotTxnId, snapshotAttempt);
                            snapshotAttempt.pathType = SnapshotPathType.SNAP_CL.toString();
                        }
                        snapshotAttempt.nonce = nonce;
                        snapshotAttempt.path = VoltDB.instance().getCommandLogSnapshotPath();
                    } finally {
                        // TRAIL [TruncSnap:9] (callback) restart the whole request check cycle
                        try {
                            truncationRequestExistenceCheck();
                        } catch (Exception e) {
                            VoltDB.crashLocalVoltDB(
                                    "Unexpected error checking for existence of truncation snapshot request"
                                    , true, e);
                        }
                    }
                } else {
                    loggingLog.info("Retrying log truncation snapshot in 60 seconds");
                    /*
                     * TRAIL [TruncSnap:10] (callback) on table reported failure try again in a few minutes
                     */
                    m_es.schedule(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                processTruncationRequestEvent(event);
                            } catch (Exception e) {
                                VoltDB.crashLocalVoltDB("Exception processing truncation request event", true, e);
                            }
                        }
                    }, 1, TimeUnit.MINUTES);
                }
            }

        });
        try {
            loggingLog.info("Initiating @SnapshotSave for log truncation");
            initiateSnapshotSave(handle, new Object[]{jsObj.toString(4)}, false);
        } catch (JSONException e) {
            /*
             * Should never happen, so fail fast
             */
            VoltDB.crashLocalVoltDB("", true, e);
        }
        return;
    }
private void processSnapshotTruncationRequestCreated ( final WatchedEvent event ) { loggingLog . info ( "Snapshot truncation leader received snapshot truncation request" ) ; final String truncReqId ; try { List < String > children = m_zk . getChildren ( event . getPath ( ) , false ) ; if ( children . isEmpty ( ) ) { loggingLog . error ( "Unable to retrieve truncation snapshot request id from ZK, log can't be truncated" ) ; return ; } truncReqId = ZKUtil . joinZKPath ( event . getPath ( ) , Collections . max ( children ) ) ; } catch ( Exception e ) { loggingLog . error ( "Unable to retrieve truncation snapshot request ID from ZK, log can't be truncated" ) ; return ; } final long now = System . currentTimeMillis ( ) ; final String nonce = Long . toString ( now ) ; try { ByteBuffer payload = ByteBuffer . allocate ( 8 ) ; payload . putLong ( 0 , now ) ; m_zk . setData ( VoltZK . request_truncation_snapshot , payload . array ( ) , - 1 ) ; } catch ( Exception e ) { VoltDB . crashLocalVoltDB ( "Setting data on the truncation snapshot request in ZK should never fail" , true , e ) ; } JSONObject jsObj = new JSONObject ( ) ; try { assert truncReqId != null ; String sData = "" ; JSONObject jsData = new JSONObject ( ) ; jsData . put ( "truncReqId" , truncReqId ) ; sData = jsData . toString ( ) ; jsObj . put ( SnapshotUtil . JSON_PATH , VoltDB . instance ( ) . getCommandLogSnapshotPath ( ) ) ; jsObj . put ( SnapshotUtil . JSON_NONCE , nonce ) ; jsObj . put ( SnapshotUtil . JSON_PATH_TYPE , SnapshotPathType . SNAP_CL ) ; jsObj . put ( "perPartitionTxnIds" , retrievePerPartitionTransactionIds ( ) ) ; jsObj . put ( "data" , sData ) ; } catch ( JSONException e ) { VoltDB . crashLocalVoltDB ( "" , true , e ) ; } long handle = m_nextCallbackHandle ++ ; m_procedureCallbacks . put ( handle , new ProcedureCallback ( ) { @ Override public void clientCallback ( ClientResponse clientResponse ) throws Exception { m_lastInitiationTs = null ; if ( clientResponse . getStatus ( ) != ClientResponse . SUCCESS ) { loggingLog . warn ( "Attempt to initiate a truncation snapshot was not successful: " + clientResponse . getStatusString ( ) ) ; loggingLog . warn ( "Retrying log truncation snapshot in 5 minutes" ) ; m_es . schedule ( new Runnable ( ) { @ Override public void run ( ) { try { processTruncationRequestEvent ( event ) ; } catch ( Exception e ) { VoltDB . crashLocalVoltDB ( "Error processing snapshot truncation request event" , true , e ) ; } } } , 5 , TimeUnit . MINUTES ) ; return ; } final VoltTable results [ ] = clientResponse . getResults ( ) ; final VoltTable result = results [ 0 ] ; boolean success = true ; final String err = SnapshotUtil . didSnapshotRequestFailWithErr ( results ) ; if ( err != null ) { if ( err . trim ( ) . equalsIgnoreCase ( "SNAPSHOT IN PROGRESS" ) ) { loggingLog . info ( "Snapshot is in progress" ) ; } else { loggingLog . warn ( "Snapshot failed with failure response: " + err ) ; } success = false ; } if ( success ) { while ( result . advanceRow ( ) ) { if ( ! result . getString ( "RESULT" ) . equals ( "SUCCESS" ) ) { success = false ; loggingLog . warn ( "Snapshot save feasibility test failed for host " + result . getLong ( "HOST_ID" ) + " table " + result . getString ( "TABLE" ) + " with error message " + result . getString ( "ERR_MSG" ) ) ; } } } if ( success ) { loggingLog . info ( "Snapshot initiation for log truncation was successful" ) ; JSONObject obj = new JSONObject ( clientResponse . getAppStatusString ( ) ) ; final long snapshotTxnId = Long . valueOf ( obj . getLong ( "txnId" ) ) ; try { boolean found = false ; ZKUtil . VoidCallback lastCallback = null ; for ( String child : m_zk . getChildren ( event . getPath ( ) , false ) ) { String requestId = ZKUtil . joinZKPath ( event . getPath ( ) , child ) ; found = found || requestId . equals ( truncReqId ) ; lastCallback = new ZKUtil . VoidCallback ( ) ; m_zk . delete ( requestId , - 1 , lastCallback , null ) ; } if ( lastCallback != null ) { try { lastCallback . get ( ) ; } catch ( KeeperException . NoNodeException ignoreIt ) { } } if ( ! found ) { VoltDB . crashLocalVoltDB ( "Could not match truncations snapshot request id while atepting its removal" , true , null ) ; } } catch ( Exception e ) { VoltDB . crashLocalVoltDB ( "Unexpected error deleting truncation snapshot request" , true , e ) ; } try { TruncationSnapshotAttempt snapshotAttempt = m_truncationSnapshotAttempts . get ( snapshotTxnId ) ; if ( snapshotAttempt == null ) { snapshotAttempt = new TruncationSnapshotAttempt ( ) ; m_truncationSnapshotAttempts . put ( snapshotTxnId , snapshotAttempt ) ; snapshotAttempt . pathType = SnapshotPathType . SNAP_CL . toString ( ) ; } snapshotAttempt . nonce = nonce ; snapshotAttempt . path = VoltDB . instance ( ) . getCommandLogSnapshotPath ( ) ; } finally { try { truncationRequestExistenceCheck ( ) ; } catch ( Exception e ) { VoltDB . crashLocalVoltDB ( "Unexpected error checking for existence of truncation snapshot request" , true , e ) ; } } } else { loggingLog . info ( "Retrying log truncation snapshot in 60 seconds" ) ; m_es . schedule ( new Runnable ( ) { @ Override public void run ( ) { try { processTruncationRequestEvent ( event ) ; } catch ( Exception e ) { VoltDB . crashLocalVoltDB ( "Exception processing truncation request event" , true , e ) ; } } } , 1 , TimeUnit . MINUTES ) ; } } } ) ; try { loggingLog . info ( "Initiating @SnapshotSave for log truncation" ) ; initiateSnapshotSave ( handle , new Object [ ] { jsObj . toString ( 4 ) } , false ) ; } catch ( JSONException e ) { VoltDB . crashLocalVoltDB ( "" , true , e ) ; } return ; }
/*
A ZK event occured requestion a truncation snapshot be taken