code: private static long getInterval(long lowWatermarkValue, long highWatermarkValue, long partitionInterval,
        int maxIntervals) {
      if (lowWatermarkValue > highWatermarkValue) {
        LOG.info(
            "lowWatermarkValue: " + lowWatermarkValue + " is greater than highWatermarkValue: " + highWatermarkValue);
  
        return 0;
      }
      long outputInterval = partitionInterval;
      boolean longOverflow = false;
      long totalIntervals = Long.MAX_VALUE;
      try {
        totalIntervals = DoubleMath.roundToLong(
            (double) highWatermarkValue / partitionInterval - (double) lowWatermarkValue / partitionInterval,
            RoundingMode.CEILING);
      } catch (java.lang.ArithmeticException e) {
        longOverflow = true;
      }
      if (longOverflow || totalIntervals > maxIntervals) {
        outputInterval = DoubleMath.roundToLong(
            (double) highWatermarkValue / maxIntervals - (double) lowWatermarkValue / maxIntervals, RoundingMode.CEILING);
  
      }
      return outputInterval;
    }
docs: recalculate interval if total number of partitions greater than maximum number of allowed partitions
  
  @param lowWatermarkValue low watermark value
  @param highWatermarkValue high watermark value
  @param partitionInterval partition interval
  @param maxIntervals max number of allowed partitions
  @return calculated interval