@VisibleForTesting
  public void updatePartitionCountMetric() {
    try {
      Map<SystemStream, SystemStreamMetadata> currentMetadata = getMetadata(streamsToMonitor, metadataCache);
      Set<SystemStream> streamsChanged = new HashSet<>();

      for (Map.Entry<SystemStream, SystemStreamMetadata> metadataEntry : initialMetadata.entrySet()) {
        try {
          SystemStream systemStream = metadataEntry.getKey();
          SystemStreamMetadata metadata = metadataEntry.getValue();

          int currentPartitionCount = currentMetadata.get(systemStream).getSystemStreamPartitionMetadata().size();
          int prevPartitionCount = metadata.getSystemStreamPartitionMetadata().size();

          Gauge gauge = gauges.get(systemStream);
          gauge.set(currentPartitionCount);
          if (currentPartitionCount != prevPartitionCount) {
            log.warn(String.format("Change of partition count detected in stream %s. old partition count: %d, current partition count: %d",
                systemStream.toString(), prevPartitionCount, currentPartitionCount));
            if (currentPartitionCount  > prevPartitionCount) {
              log.error(String.format("Shutting down (stateful) or restarting (stateless) the job since current " +
                      "partition count %d is greater than the old partition count %d for stream %s.",
                  currentPartitionCount, prevPartitionCount, systemStream.toString()));
              streamsChanged.add(systemStream);
            }
          }
        } catch (Exception e) {
          log.error(String.format("Error comparing partition count differences for stream: %s", metadataEntry.getKey().toString()));
        }
      }

      if (!streamsChanged.isEmpty() && this.callbackMethod != null) {
        this.callbackMethod.onSystemStreamPartitionChange(streamsChanged);
      }

    } catch (Exception e) {
      log.error("Exception while updating partition count metric.", e);
    }
  }
Fetches the current partition count for each system stream from the cache, compares the current count to the
original count and updates the metric for that system stream with the delta.
samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitor.java