TaskSideInputHandler { @VisibleForTesting Map<SystemStreamPartition, String> getStartingOffsets( Map<SystemStreamPartition, String> fileOffsets, Map<SystemStreamPartition, String> oldestOffsets) { Map<SystemStreamPartition, String> startingOffsets = new HashMap<>(); this.sspToStores.keySet().forEach(ssp -> { String fileOffset = fileOffsets.get(ssp); String oldestOffset = oldestOffsets.get(ssp); startingOffsets.put(ssp, this.storageManagerUtil.getStartingOffset( ssp, this.systemAdmins.getSystemAdmin(ssp.getSystem()), fileOffset, oldestOffset)); }); return startingOffsets; } TaskSideInputHandler(TaskName taskName, TaskMode taskMode, File storeBaseDir,
      Map<String, StorageEngine> storeToStorageEngines, Map<String, Set<SystemStreamPartition>> storeToSSPs,
      Map<String, SideInputsProcessor> storeToProcessor, SystemAdmins systemAdmins,
      StreamMetadataCache streamMetadataCache, CountDownLatch taskCaughtUpLatch, Clock clock); TaskName getTaskName(); void init(); synchronized void process(IncomingMessageEnvelope envelope); synchronized void flush(); String getStartingOffset(SystemStreamPartition ssp); String getLastProcessedOffset(SystemStreamPartition ssp); void stop();  }
@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(); 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 ? getOffsetAfter(String.valueOf(ssp.getPartition().getPartitionId() + 10)) : String.valueOf(ssp.getPartition().getPartitionId() + 10); assertEquals("Larger of fileOffsets and oldestOffsets should always be chosen", expectedOffset, offset); }); }