@Test
  public void testGetStartingOffsets() {
    final String storeName = "test-get-starting-offset-store";
    final String taskName = "test-get-starting-offset-task";

    Set<SystemStreamPartition> ssps = IntStream.range(1, 6)
        .mapToObj(idx -> new SystemStreamPartition(TEST_SYSTEM, TEST_STREAM, new Partition(idx)))
        .collect(Collectors.toSet());


    TaskSideInputHandler handler = new MockTaskSideInputHandlerBuilder(taskName, TaskMode.Active)
        .addStore(storeName, ssps)
        .build();

    // set up file and oldest offsets. for even partitions, fileOffsets will be larger; for odd partitions oldestOffsets will be larger
    Map<SystemStreamPartition, String> fileOffsets = ssps.stream()
        .collect(Collectors.toMap(Function.identity(), ssp -> {
          int partitionId = ssp.getPartition().getPartitionId();
          int offset = partitionId % 2 == 0 ? partitionId + 10 : partitionId;
          return String.valueOf(offset);
        }));
    Map<SystemStreamPartition, String> oldestOffsets = ssps.stream()
        .collect(Collectors.toMap(Function.identity(), ssp -> {
          int partitionId = ssp.getPartition().getPartitionId();
          int offset = partitionId % 2 == 0 ? partitionId : partitionId + 10;

          return String.valueOf(offset);
        }));

    doCallRealMethod().when(handler).getStartingOffsets(fileOffsets, oldestOffsets);

    Map<SystemStreamPartition, String> startingOffsets = handler.getStartingOffsets(fileOffsets, oldestOffsets);

    assertTrue("Failed to get starting offsets for all ssps", startingOffsets.size() == 5);
    startingOffsets.forEach((ssp, offset) -> {
      int partitionId = ssp.getPartition().getPartitionId();
      String expectedOffset = partitionId % 2 == 0
          // 1 + fileOffset
          ? getOffsetAfter(String.valueOf(ssp.getPartition().getPartitionId() + 10))
          // oldestOffset
          : String.valueOf(ssp.getPartition().getPartitionId() + 10);
      assertEquals("Larger of fileOffsets and oldestOffsets should always be chosen", expectedOffset, offset);
    });
  }
This test is for cases, when calls to systemAdmin (e.g., KafkaSystemAdmin's) get-stream-metadata method return null.
samza-core/src/test/java/org/apache/samza/storage/TestTaskSideInputHandler.java