docs: @SuppressWarnings({ "unchecked", "rawtypes", "unused" })
      private void commitToVoldemort(List<String> storeNamesToCommit) {
  
          if(logger.isDebugEnabled()) {
              logger.debug("Trying to commit to Voldemort");
          }
  
          boolean hasError = false;
          if(nodesToStream == null || nodesToStream.size() == 0) {
              if(logger.isDebugEnabled()) {
                  logger.debug("No nodes to stream to. Returning.");
              }
              return;
          }
  
          for(Node node: nodesToStream) {
  
              for(String store: storeNamesToCommit) {
                  if(!nodeIdStoreInitialized.get(new Pair(store, node.getId())))
                      continue;
  
                  nodeIdStoreInitialized.put(new Pair(store, node.getId()), false);
  
                  DataOutputStream outputStream = nodeIdStoreToOutputStreamRequest.get(new Pair(store,
                                                                                                node.getId()));
  
                  try {
                      ProtoUtils.writeEndOfStream(outputStream);
                      outputStream.flush();
                      DataInputStream inputStream = nodeIdStoreToInputStreamRequest.get(new Pair(store,
                                                                                                 node.getId()));
                      VAdminProto.UpdatePartitionEntriesResponse.Builder updateResponse = ProtoUtils.readToBuilder(inputStream,
                                                                                                                   VAdminProto.UpdatePartitionEntriesResponse.newBuilder());
                      if(updateResponse.hasError()) {
                          hasError = true;
                      }
  
                  } catch(IOException e) {
                      logger.error("Exception during commit", e);
                      hasError = true;
                      if(!faultyNodes.contains(node.getId()))
                          faultyNodes.add(node.getId());
                  }
              }
  
          }
  
          if(streamingresults == null) {
              logger.warn("StreamingSession may not have been initialized since Variable streamingresults is null. Skipping callback ");
              return;
          }
  
          // remove redundant callbacks
          if(hasError) {
  
              logger.info("Invoking the Recovery Callback");
              Future future = streamingresults.submit(recoveryCallback);
              try {
                  future.get();
  
              } catch(InterruptedException e1) {
                  MARKED_BAD = true;
                  logger.error("Recovery Callback failed", e1);
                  throw new VoldemortException("Recovery Callback failed");
              } catch(ExecutionException e1) {
                  MARKED_BAD = true;
                  logger.error("Recovery Callback failed during execution", e1);
                  throw new VoldemortException("Recovery Callback failed during execution");
              }
          } else {
              if(logger.isDebugEnabled()) {
                  logger.debug("Commit successful");
                  logger.debug("calling checkpoint callback");
              }
              Future future = streamingresults.submit(checkpointCallback);
              try {
                  future.get();
  
              } catch(InterruptedException e1) {
                  logger.warn("Checkpoint callback failed!", e1);
              } catch(ExecutionException e1) {
                  logger.warn("Checkpoint callback failed during execution!", e1);
              }
          }
  
      }