TransactionalStateTaskRestoreManager implements TaskRestoreManager { @VisibleForTesting static void registerStartingOffsets( TaskModel taskModel, StoreActions storeActions, Map<String, SystemStream> storeChangelogs, SystemAdmins systemAdmins, Map<String, SystemConsumer> storeConsumers, Map<SystemStreamPartition, SystemStreamPartitionMetadata> currentChangelogOffsets) { Map<String, RestoreOffsets> storesToRestore = storeActions.storesToRestore; storeChangelogs.forEach((storeName, changelog) -> { SystemStreamPartition changelogSSP = new SystemStreamPartition(changelog, taskModel.getChangelogPartition()); SystemConsumer systemConsumer = storeConsumers.get(storeName); SystemStreamPartitionMetadata currentOffsets = currentChangelogOffsets.get(changelogSSP); String upcomingOffset = currentOffsets.getUpcomingOffset(); LOG.info("Temporarily registering upcoming offset: {} as the starting offest for changelog ssp: {}. " + "This might be overridden later for stores that need restoring.", upcomingOffset, changelogSSP); systemConsumer.register(changelogSSP, upcomingOffset); }); storesToRestore.forEach((storeName, restoreOffsets) -> { SystemStream changelog = storeChangelogs.get(storeName); SystemStreamPartition changelogSSP = new SystemStreamPartition(changelog, taskModel.getChangelogPartition()); SystemAdmin systemAdmin = systemAdmins.getSystemAdmin(changelog.getSystem()); validateRestoreOffsets(restoreOffsets, systemAdmin); SystemConsumer systemConsumer = storeConsumers.get(storeName); SystemStreamPartitionMetadata currentOffsets = currentChangelogOffsets.get(changelogSSP); String oldestOffset = currentOffsets.getOldestOffset(); String startingOffset; if (systemAdmin.offsetComparator(restoreOffsets.startingOffset, oldestOffset) == 0) { startingOffset = oldestOffset; } else { Map<SystemStreamPartition, String> offsetMap = ImmutableMap.of(changelogSSP, restoreOffsets.startingOffset); startingOffset = systemAdmin.getOffsetsAfter(offsetMap).get(changelogSSP); } LOG.info("Registering starting offset: {} for changelog ssp: {}", startingOffset, changelogSSP); systemConsumer.register(changelogSSP, startingOffset); }); } TransactionalStateTaskRestoreManager(
      TaskModel taskModel,
      Map<String, StorageEngine> storeEngines,
      Map<String, SystemStream> storeChangelogs,
      SystemAdmins systemAdmins,
      Map<String, SystemConsumer> storeConsumers,
      SSPMetadataCache sspMetadataCache,
      File loggedStoreBaseDirectory,
      File nonLoggedStoreBaseDirectory,
      Config config,
      Clock clock); @Override void init(Map<SystemStreamPartition, String> checkpointedChangelogOffsets); @Override void restore(); void stopPersistentStores();  }
@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"; Map<String, RestoreOffsets> mockRestoreOffsets = ImmutableMap.of( store1Name, new RestoreOffsets("0", "5"), store2Name, new RestoreOffsets("15", "20"), store4Name, new RestoreOffsets("31", null)); 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(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"); verify(mockSystemConsumer, times(1)).register(changelog1SSP, "0"); verify(mockSystemConsumer, times(1)).register(changelog2SSP, "16"); verify(mockSystemConsumer, times(1)).register(changelog4SSP, "31"); verifyNoMoreInteractions(mockSystemConsumer); }
@Test(expected = IllegalStateException.class) public void testRegisterStartingOffsetsThrowsIfStartingGreaterThanEnding() { 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); Map<String, RestoreOffsets> mockRestoreOffsets = ImmutableMap.of("store1", new RestoreOffsets("5", "0")); StoreActions mockStoreActions = new StoreActions(ImmutableMap.of(), ArrayListMultimap.create(), mockRestoreOffsets); String store1Name = "store1"; String changelogSystemName = "system"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelogSystemName, changelog1StreamName); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogSSPMetadata = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); 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("store1", mockSystemConsumer); TransactionalStateTaskRestoreManager.registerStartingOffsets( mockTaskModel, mockStoreActions, mockStoreChangelogs, mockSystemAdmins, mockStoreConsumers, mockCurrentChangelogSSPMetadata); fail("Should have thrown an exception since starting offset > ending offset"); }