code: @SuppressWarnings("unchecked")
      private void invokePartitionChangeListener() {
          if (!triggerListeners.isEmpty()) {
              Map<PartitionEntry, Integer> current = (Map<PartitionEntry, Integer>) partitionCache.get(CURRENT);
              Map<PartitionEntry, Integer> previous = (Map<PartitionEntry, Integer>) partitionCache.get(PREVIOUS);
  
              Map<String, List<String>> partition = getNodePartition(current, currentNode);
              Map<String, Map<String, List<String>>> addedRemoved =
                      getAddedRemovedPartition(previous, current, currentNode);
              if (log.isDebugEnabled()) {
                  log.debug("Invoke a Change Listener");
                  log.debugf("Previous: %s", previous);
                  log.debugf("Current: %s", current);
                  log.debugf("Partition: %s", partition);
                  log.debugf("Added: %s", addedRemoved.get("added"));
                  log.debugf("Removed: %s", addedRemoved.get("removed"));
              }
              triggerListeners.stream().forEach(triggerListener -> {
                  triggerListener.onPartitionChange(partition, addedRemoved.get("removed"), addedRemoved.get("added"));
              });
          }
      }
docs: /*
  Invoke PartitionTriggerListener with local, added and removed partition