docs: @SuppressWarnings({"checkstyle:npathcomplexity", "UnnecessaryBoxing"})
      @SuppressFBWarnings(value = "DM_NUMBER_CTOR", justification = "we need a shared counter object for each member per partition")
      protected void putAllInternal(Map<?, ?> map) {
          try {
              int mapSize = map.size();
              if (mapSize == 0) {
                  return;
              }
  
              boolean useBatching = isPutAllUseBatching(mapSize);
              int partitionCount = partitionService.getPartitionCount();
              int initialSize = getPutAllInitialSize(useBatching, mapSize, partitionCount);
  
              Map<Address, List<Integer>> memberPartitionsMap = partitionService.getMemberPartitionsMap();
  
              // init counters for batching
              MutableLong[] counterPerMember = null;
              Address[] addresses = null;
              if (useBatching) {
                  counterPerMember = new MutableLong[partitionCount];
                  addresses = new Address[partitionCount];
                  for (Entry<Address, List<Integer>> addressListEntry : memberPartitionsMap.entrySet()) {
                      MutableLong counter = new MutableLong();
                      Address address = addressListEntry.getKey();
                      for (int partitionId : addressListEntry.getValue()) {
                          counterPerMember[partitionId] = counter;
                          addresses[partitionId] = address;
                      }
                  }
              }
  
              // fill entriesPerPartition
              MapEntries[] entriesPerPartition = new MapEntries[partitionCount];
              for (Entry entry : map.entrySet()) {
                  checkNotNull(entry.getKey(), NULL_KEY_IS_NOT_ALLOWED);
                  checkNotNull(entry.getValue(), NULL_VALUE_IS_NOT_ALLOWED);
  
                  Data keyData = toDataWithStrategy(entry.getKey());
                  int partitionId = partitionService.getPartitionId(keyData);
                  MapEntries entries = entriesPerPartition[partitionId];
                  if (entries == null) {
                      entries = new MapEntries(initialSize);
                      entriesPerPartition[partitionId] = entries;
                  }
  
                  entries.add(keyData, toData(entry.getValue()));
  
                  if (useBatching) {
                      long currentSize = ++counterPerMember[partitionId].value;
                      if (currentSize % putAllBatchSize == 0) {
                          List<Integer> partitions = memberPartitionsMap.get(addresses[partitionId]);
                          invokePutAllOperation(partitions, entriesPerPartition);
                      }
                  }
              }
  
              // invoke operations for entriesPerPartition
              for (Entry<Address, List<Integer>> entry : memberPartitionsMap.entrySet()) {
                  invokePutAllOperation(entry.getValue(), entriesPerPartition);
              }
  
              finalizePutAll(map);
          } catch (Exception e) {
              throw rethrow(e);
          }
      }