docs: @SuppressWarnings({ "unchecked", "rawtypes" })
      public synchronized void streamingPut(ByteArray key, Versioned<byte[]> value, String storeName) {
  
          // If store does not exist in the stores list
          // add it and checkout a socket
          if(!storeNames.contains(storeName)) {
              addStoreToSession(storeName);
          }
  
          if(MARKED_BAD) {
              logger.error("Cannot stream more entries since Recovery Callback Failed!");
              throw new VoldemortException("Cannot stream more entries since Recovery Callback Failed! You Need to restart the session");
          }
  
          List<Node> nodeList = storeToRoutingStrategy.get(storeName).routeRequest(key.get());
  
          int nodesWithException = 0;
          // sent the k/v pair to the nodes
          for(Node node: nodeList) {
  
              if(blackListedNodes != null && blackListedNodes.size() > 0) {
                  if(blackListedNodes.contains(node.getId()))
                      continue;
              }
              // if node! in blacklistednodes
  
              VAdminProto.PartitionEntry partitionEntry = VAdminProto.PartitionEntry.newBuilder()
                                                                                    .setKey(ProtoUtils.encodeBytes(key))
                                                                                    .setVersioned(ProtoUtils.encodeVersioned(value))
                                                                                    .build();
  
              VAdminProto.UpdatePartitionEntriesRequest.Builder updateRequest = null;
  
              if(overWriteIfLatestTs) {
                  updateRequest = VAdminProto.UpdatePartitionEntriesRequest.newBuilder()
                                                                           .setStore(storeName)
                                                                           .setPartitionEntry(partitionEntry)
                                                                           .setOverwriteIfLatestTs(overWriteIfLatestTs);
              } else {
                  updateRequest = VAdminProto.UpdatePartitionEntriesRequest.newBuilder()
                                                                           .setStore(storeName)
                                                                           .setPartitionEntry(partitionEntry);
              }
  
              DataOutputStream outputStream = nodeIdStoreToOutputStreamRequest.get(new Pair(storeName,
                                                                                            node.getId()));
              try {
                  if(nodeIdStoreInitialized.get(new Pair(storeName, node.getId()))) {
                      ProtoUtils.writeMessage(outputStream, updateRequest.build());
                  } else {
                      ProtoUtils.writeMessage(outputStream,
                                              VAdminProto.VoldemortAdminRequest.newBuilder()
                                                                               .setType(VAdminProto.AdminRequestType.UPDATE_PARTITION_ENTRIES)
                                                                               .setUpdatePartitionEntries(updateRequest)
                                                                               .build());
                      outputStream.flush();
                      nodeIdStoreInitialized.put(new Pair(storeName, node.getId()), true);
  
                  }
  
              } catch(IOException e) {
                  e.printStackTrace();
                  nodesWithException++;
                  if(!faultyNodes.contains(node.getId()))
                      faultyNodes.add(node.getId());
              }
  
          }
  
          if(nodesWithException > 0) {
              logger.warn("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 {
              entriesProcessed++;
              if(entriesProcessed == CHECKPOINT_COMMIT_SIZE) {
                  entriesProcessed = 0;
                  commitToVoldemort();
              }
  
              throttler.maybeThrottle(1);
          }
  
      }