@Test
  public void testRegisterStartingOffsets() {
    TaskModel mockTaskModel = mock(TaskModel.class);
    TaskName taskName = new TaskName("Partition 0");
    when(mockTaskModel.getTaskName()).thenReturn(taskName);
    Partition taskChangelogPartition = new Partition(0);
    when(mockTaskModel.getChangelogPartition()).thenReturn(taskChangelogPartition);

    String store1Name = "store1";
    String changelogSystemName = "system";
    String changelog1StreamName = "store1Changelog";
    String store2Name = "store2";
    String changelog2StreamName = "store2Changelog";
    String store3Name = "store3";
    String changelog3StreamName = "store3Changelog";
    String store4Name = "store4";
    String changelog4StreamName = "store4Changelog";

    // tests restore for store 1 and store 2 but not store 3
    Map<String, RestoreOffsets> mockRestoreOffsets = ImmutableMap.of(
        store1Name, new RestoreOffsets("0", "5"), // tests starting offset == oldest (i.e. restore is inclusive)
        store2Name, new RestoreOffsets("15", "20"), // tests starting offset != oldest (i.e. restore from next offset)
        store4Name, new RestoreOffsets("31", null)); // tests that null ending offsets are OK (should trim)
    StoreActions mockStoreActions = new StoreActions(ImmutableMap.of(), ArrayListMultimap.create(), mockRestoreOffsets);


    SystemStream changelog1SystemStream = new SystemStream(changelogSystemName, changelog1StreamName);
    SystemStream changelog2SystemStream = new SystemStream(changelogSystemName, changelog2StreamName);
    SystemStream changelog3SystemStream = new SystemStream(changelogSystemName, changelog3StreamName);
    SystemStream changelog4SystemStream = new SystemStream(changelogSystemName, changelog4StreamName);

    Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(
        store1Name, changelog1SystemStream,
        store2Name, changelog2SystemStream,
        store3Name, changelog3SystemStream,
        store4Name, changelog4SystemStream);

    SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition);
    SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11");
    SystemStreamPartition changelog2SSP = new SystemStreamPartition(changelog2SystemStream, taskChangelogPartition);
    SystemStreamPartitionMetadata changelog2SSPMetadata = new SystemStreamPartitionMetadata("11", "20", "21");
    SystemStreamPartition changelog3SSP = new SystemStreamPartition(changelog3SystemStream, taskChangelogPartition);
    SystemStreamPartitionMetadata changelog3SSPMetadata = new SystemStreamPartitionMetadata("21", "30", "31");
    SystemStreamPartition changelog4SSP = new SystemStreamPartition(changelog4SystemStream, taskChangelogPartition);
    SystemStreamPartitionMetadata changelog4SSPMetadata = new SystemStreamPartitionMetadata("31", "40", "41");
    Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogSSPMetadata =
        ImmutableMap.of(changelog1SSP, changelog1SSPMetadata,
            changelog2SSP, changelog2SSPMetadata,
            changelog3SSP, changelog3SSPMetadata,
            changelog4SSP, changelog4SSPMetadata);

    SystemAdmins mockSystemAdmins = mock(SystemAdmins.class);
    SystemAdmin mockSystemAdmin = mock(SystemAdmin.class);
    when(mockSystemAdmins.getSystemAdmin(eq(changelogSystemName))).thenReturn(mockSystemAdmin);
    Mockito.when(mockSystemAdmin.offsetComparator(anyString(), anyString()))
        .thenAnswer((Answer<Integer>) invocation -> {
          String offset1 = (String) invocation.getArguments()[0];
          String offset2 = (String) invocation.getArguments()[1];
          return Long.valueOf(offset1).compareTo(Long.valueOf(offset2));
        });
    Mockito.when(mockSystemAdmin.getOffsetsAfter(any()))
        .thenAnswer((Answer<Map<SystemStreamPartition, String>>) invocation -> {
          Map<SystemStreamPartition, String> offsets = (Map<SystemStreamPartition, String>) invocation.getArguments()[0];
          Map<SystemStreamPartition, String> nextOffsets = new HashMap<>();
          offsets.forEach((ssp, offset) -> nextOffsets.put(ssp, Long.toString(Long.valueOf(offset) + 1)));
          return nextOffsets;
        });

    SystemConsumer mockSystemConsumer = mock(SystemConsumer.class);
    Map<String, SystemConsumer> mockStoreConsumers = ImmutableMap.of(
        store1Name, mockSystemConsumer,
        store2Name, mockSystemConsumer,
        store3Name, mockSystemConsumer,
        store4Name, mockSystemConsumer);

    TransactionalStateTaskRestoreManager.registerStartingOffsets(
        mockTaskModel, mockStoreActions, mockStoreChangelogs, mockSystemAdmins,
        mockStoreConsumers, mockCurrentChangelogSSPMetadata);

    // verify that we first register upcoming offsets for each changelog ssp
    verify(mockSystemConsumer, times(1)).register(changelog1SSP, "11");
    verify(mockSystemConsumer, times(1)).register(changelog2SSP, "21");
    verify(mockSystemConsumer, times(1)).register(changelog3SSP, "31");
    verify(mockSystemConsumer, times(1)).register(changelog4SSP, "41");

    // then verify that we override the starting offsets for changelog 1 and 2
    verify(mockSystemConsumer, times(1)).register(changelog1SSP, "0"); // ensure that starting offset is inclusive if oldest
    verify(mockSystemConsumer, times(1)).register(changelog2SSP, "16"); // and that it is next offset if not oldest
    verify(mockSystemConsumer, times(1)).register(changelog4SSP, "31"); // and that null ending offset is ok
    verifyNoMoreInteractions(mockSystemConsumer);
  }
This can happen if the changelog topic was manually deleted and recreated, and the checkpointed/local changelog
offset is not valid anymore.
samza-core/src/test/java/org/apache/samza/storage/TestTransactionalStateTaskRestoreManager.java