code: @SuppressWarnings("unchecked")
      protected Map<String, Map<String, List<String>>> getAddedRemovedPartition(Map<PartitionEntry, Integer> previous,
                                                                              Map<PartitionEntry, Integer> current,
                                                                              Integer node) {
          Map<String, Map<String, List<String>>> output = new HashMap<>();
          output.put("added", new HashMap<>());
          output.put("removed", new HashMap<>());
  
          if (isEmpty(previous)) {
              current.entrySet().stream().forEach(entry -> {
                  add(output.get("added"), entry.getKey());
              });
          } else {
              List<PartitionEntry> previousNode = new ArrayList();
              for (Entry<PartitionEntry, Integer> entry : previous.entrySet()) {
                  if (entry.getValue().equals(node)) {
                      previousNode.add(entry.getKey());
                  }
              }
              List<PartitionEntry> currentNode = new ArrayList();
              for (Entry<PartitionEntry, Integer> entry : current.entrySet()) {
                  if (entry.getValue().equals(node)) {
                      currentNode.add(entry.getKey());
                  }
              }
              for (PartitionEntry entry : previousNode) {
                  if (!currentNode.contains(entry)) {
                      add(output.get("removed"), entry);
                  }
              }
              for (PartitionEntry entry : currentNode) {
                  if (!previousNode.contains(entry)) {
                      add(output.get("added"), entry);
                  }
              }
          }
          return output;
      }
docs: /*
  Calculated the added and removed entries for a node given a current and a previous partition maps.
  It return a map with two fixed entries under keys "added" and "removed".