private void checkCaughtUp(SystemStreamPartition ssp, String currentOffset, SystemStreamMetadata.OffsetType offsetTypeToCheck) {
    SystemStreamMetadata.SystemStreamPartitionMetadata sspMetadata = this.initialSideInputSSPMetadata.get(ssp);
    String offsetToCheck = sspMetadata == null ? null : sspMetadata.getOffset(offsetTypeToCheck);

    LOG.trace("Checking offset {} against {} offset {} for {}.", currentOffset, offsetToCheck, offsetTypeToCheck, ssp);

    Integer comparatorResult;
    if (currentOffset == null || offsetToCheck == null) {
      comparatorResult = -1;
    } else {
      SystemAdmin systemAdmin = systemAdmins.getSystemAdmin(ssp.getSystem());
      comparatorResult = systemAdmin.offsetComparator(currentOffset, offsetToCheck);
    }

    // The SSP is no longer lagging if the envelope's offset is greater than or equal to the
    // latest offset.
    if (comparatorResult != null && comparatorResult.intValue() >= 0) {
      LOG.info("Side input ssp {} has caught up to offset {}.", ssp, offsetToCheck);
      // if its caught up, we remove the ssp from the map
      this.initialSideInputSSPMetadata.remove(ssp);
      if (this.initialSideInputSSPMetadata.isEmpty()) {
        // if the metadata list is now empty, all SSPs in the task are caught up so count down the latch
        // this will only happen once, when the last ssp catches up
        this.taskCaughtUpLatch.countDown();
      }
    }
  }
An SSP is considered caught up once the offset indicated for it in {@link #initialSideInputSSPMetadata} has been
processed. Once the set of SSPs to catch up becomes empty, the latch for the task will count down, notifying
{@link ContainerStorageManager} that it is caught up.

@param ssp The SSP to be checked
@param currentOffset The offset to be checked
@param offsetTypeToCheck The type offset to compare {@code currentOffset} to.
samza-core/src/main/java/org/apache/samza/storage/TaskSideInputHandler.java