try {
    executeStopWatch.start();
    changeEventSink.execute(records);
    executeStopWatch.stop();
    
}
catch (Throwable throwable) {

    // Capture failure
    LOGGER.error("Failed to process record: {}", throwable.getMessage(), throwable);
   
}

putStopWatch.stop();
LOGGER.trace("[PERF] Total put execution time {}", putStopWatch.durations());
LOGGER.trace("[PERF] Sink execute execution time {}", executeStopWatch.durations());