TransactionalStateTaskRestoreManager implements TaskRestoreManager { @VisibleForTesting static StoreActions getStoreActions( TaskModel taskModel, Map<String, StorageEngine> storeEngines, Map<String, SystemStream> storeChangelogs, Map<SystemStreamPartition, String> checkpointedChangelogOffsets, Map<SystemStreamPartition, SystemStreamPartitionMetadata> currentChangelogOffsets, SystemAdmins systemAdmins, StorageManagerUtil storageManagerUtil, File loggedStoreBaseDirectory, File nonLoggedStoreBaseDirectory, Config config, Clock clock) { TaskName taskName = taskModel.getTaskName(); TaskMode taskMode = taskModel.getTaskMode(); Map<String, File> storeDirToRetain = new HashMap<>(); ListMultimap<String, File> storeDirsToDelete = ArrayListMultimap.create(); Map<String, RestoreOffsets> storesToRestore = new HashMap<>(); storeEngines.forEach((storeName, storageEngine) -> { if (!storageEngine.getStoreProperties().isPersistedToDisk() && !storageEngine.getStoreProperties().isLoggedStore()) { return; } if (storageEngine.getStoreProperties().isPersistedToDisk() && !storageEngine.getStoreProperties().isLoggedStore()) { File currentDir = storageManagerUtil.getTaskStoreDir( nonLoggedStoreBaseDirectory, storeName, taskName, taskMode); LOG.info("Marking current directory: {} for store: {} in task: {} for deletion since it is not a logged store.", currentDir, storeName, taskName); storeDirsToDelete.put(storeName, currentDir); return; } SystemStream changelog = storeChangelogs.get(storeName); SystemStreamPartition changelogSSP = new SystemStreamPartition(changelog, taskModel.getChangelogPartition()); SystemAdmin admin = systemAdmins.getSystemAdmin(changelogSSP.getSystem()); SystemStreamPartitionMetadata changelogSSPMetadata = currentChangelogOffsets.get(changelogSSP); String oldestOffset = changelogSSPMetadata.getOldestOffset(); String newestOffset = changelogSSPMetadata.getNewestOffset(); String checkpointMessage = checkpointedChangelogOffsets.get(changelogSSP); String checkpointedOffset = null; long timeSinceLastCheckpointInMs = Long.MAX_VALUE; if (StringUtils.isNotBlank(checkpointMessage)) { CheckpointedChangelogOffset checkpointedChangelogOffset = CheckpointedChangelogOffset.fromString(checkpointMessage); checkpointedOffset = checkpointedChangelogOffset.getOffset(); timeSinceLastCheckpointInMs = System.currentTimeMillis() - checkpointedChangelogOffset.getCheckpointId().getMillis(); } if (storageEngine.getStoreProperties().isPersistedToDisk() && new StorageConfig( config).getCleanLoggedStoreDirsOnStart(storeName)) { File currentDir = storageManagerUtil.getTaskStoreDir(loggedStoreBaseDirectory, storeName, taskName, taskMode); LOG.info("Marking current directory: {} for store: {} in task: {} for deletion due to clean.on.container.start config.", currentDir, storeName, taskName); storeDirsToDelete.put(storeName, currentDir); storageManagerUtil.getTaskStoreCheckpointDirs(loggedStoreBaseDirectory, storeName, taskName, taskMode) .forEach(checkpointDir -> { LOG.info("Marking checkpoint directory: {} for store: {} in task: {} for deletion due to clean.on.container.start config.", checkpointDir, storeName, taskName); storeDirsToDelete.put(storeName, checkpointDir); }); LOG.info("Marking restore offsets for store: {} in task: {} to {}, {} ", storeName, taskName, oldestOffset, checkpointedOffset); storesToRestore.put(storeName, new RestoreOffsets(oldestOffset, checkpointedOffset)); return; } Optional<File> currentDirOptional; Optional<List<File>> checkpointDirsOptional; if (!storageEngine.getStoreProperties().isPersistedToDisk()) { currentDirOptional = Optional.empty(); checkpointDirsOptional = Optional.empty(); } else { currentDirOptional = Optional.of(storageManagerUtil.getTaskStoreDir( loggedStoreBaseDirectory, storeName, taskName, taskMode)); checkpointDirsOptional = Optional.of(storageManagerUtil.getTaskStoreCheckpointDirs( loggedStoreBaseDirectory, storeName, taskName, taskMode)); } LOG.info("For store: {} in task: {} got current dir: {}, checkpoint dirs: {}, checkpointed changelog offset: {}", storeName, taskName, currentDirOptional, checkpointDirsOptional, checkpointedOffset); currentDirOptional.ifPresent(currentDir -> { LOG.info("Marking current directory: {} for store: {} in task: {} for deletion.", currentDir, storeName, taskName); storeDirsToDelete.put(storeName, currentDir); }); if (checkpointedOffset == null && oldestOffset != null) { checkpointDirsOptional.ifPresent(checkpointDirs -> checkpointDirs.forEach(checkpointDir -> { LOG.info("Marking checkpoint directory: {} for store: {} in task: {} for deletion since checkpointed " + "offset is null and oldest offset: {} is not.", checkpointDir, storeName, taskName, oldestOffset); storeDirsToDelete.put(storeName, checkpointDir); })); if (new TaskConfig(config).getTransactionalStateRetainExistingState()) { LOG.warn("Checkpointed offset for store: {} in task: {} is null. Since retain existing state is true, " + "local state will be fully restored from current changelog contents. " + "There is no transactional local state guarantee.", storeName, taskName); storesToRestore.put(storeName, new RestoreOffsets(oldestOffset, newestOffset)); } else { LOG.warn("Checkpointed offset for store: {} in task: {} is null. Since retain existing state is false, " + "any local state and changelog topic contents will be deleted.", storeName, taskName); storesToRestore.put(storeName, new RestoreOffsets(oldestOffset, null)); } } else if ( admin.offsetComparator(oldestOffset, checkpointedOffset) > 0 || admin.offsetComparator(checkpointedOffset, newestOffset) > 0) { LOG.warn("Checkpointed offset: {} for store: {} in task: {} is out of range of oldest: {} or newest: {} offset." + "Deleting existing store and fully restoring from changelog topic from oldest to newest offset. If the topic " + "has time-based retention, there is no transactional local state guarantees. If the topic was changed," + "local state will be cleaned up and fully restored to match the new topic contents.", checkpointedOffset, storeName, taskName, oldestOffset, newestOffset); checkpointDirsOptional.ifPresent(checkpointDirs -> checkpointDirs.forEach(checkpointDir -> storeDirsToDelete.put(storeName, checkpointDir))); storesToRestore.put(storeName, new RestoreOffsets(oldestOffset, newestOffset)); } else { if (!checkpointDirsOptional.isPresent()) { LOG.info("Did not find any checkpoint directories for logged (maybe non-persistent) store: {}. Local state " + "will be fully restored from current changelog contents.", storeName); storesToRestore.put(storeName, new RestoreOffsets(oldestOffset, checkpointedOffset)); } else { String targetOffset; long minCompactionLagMs = new StorageConfig(config).getChangelogMinCompactionLagMs(storeName); if (timeSinceLastCheckpointInMs > .9 * minCompactionLagMs) { LOG.warn("Checkpointed offset for store: {} in task: {} is: {}. It is in range of oldest: {} and " + "newest: {} changelog offset. However, time since last checkpoint is: {}, which is greater than " + "0.9 * min.compaction.lag.ms: {} for the changelog topic. Since there is a chance that" + "the changelog topic has been compacted, restoring store to the end of the current changelog contents." + "There is no transactional local state guarantee.", storeName, taskName, checkpointedOffset, oldestOffset, newestOffset, timeSinceLastCheckpointInMs, minCompactionLagMs); targetOffset = newestOffset; } else { targetOffset = checkpointedOffset; } boolean hasValidCheckpointDir = false; for (File checkpointDir: checkpointDirsOptional.get()) { if (storageManagerUtil.isLoggedStoreValid( storeName, checkpointDir, config, storeChangelogs, taskModel, clock, storeEngines)) { String localOffset = storageManagerUtil.readOffsetFile( checkpointDir, Collections.singleton(changelogSSP), false).get(changelogSSP); LOG.info("Read local offset: {} for store: {} checkpoint dir: {} in task: {}", localOffset, storeName, checkpointDir, taskName); if (admin.offsetComparator(localOffset, oldestOffset) >= 0 && admin.offsetComparator(localOffset, targetOffset) <= 0 && (storesToRestore.get(storeName) == null || admin.offsetComparator(localOffset, storesToRestore.get(storeName).startingOffset) > 0)) { hasValidCheckpointDir = true; LOG.info("Temporarily marking checkpoint dir: {} for store: {} in task: {} for retention. " + "May be overridden later.", checkpointDir, storeName, taskName); storeDirToRetain.put(storeName, checkpointDir); LOG.info("Temporarily marking store: {} in task: {} for restore from beginning offset: {} to " + "ending offset: {}. May be overridden later", storeName, taskName, localOffset, targetOffset); storesToRestore.put(storeName, new RestoreOffsets(localOffset, targetOffset)); } } } for (File checkpointDir: checkpointDirsOptional.get()) { if (storeDirToRetain.get(storeName) == null || !storeDirToRetain.get(storeName).equals(checkpointDir)) { LOG.info("Marking checkpoint directory: {} for store: {} in task: {} for deletion since it is not " + "marked for retention.", checkpointDir, storeName, taskName); storeDirsToDelete.put(storeName, checkpointDir); } } if (!hasValidCheckpointDir) { storesToRestore.put(storeName, new RestoreOffsets(oldestOffset, targetOffset)); } } } }); LOG.info("Store directories to be retained in Task: {} are: {}", taskName, storeDirToRetain); LOG.info("Store directories to be deleted in Task: {} are: {}", taskName, storeDirsToDelete); LOG.info("Stores to be restored in Task: {} are: {}", taskName, storesToRestore); return new StoreActions(storeDirToRetain, storeDirsToDelete, storesToRestore); } 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 testGetStoreActionsForNonLoggedPersistentStore_AlwaysClearStore() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(false); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = ImmutableMap.of(); Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); when(mockStorageManagerUtil.getTaskStoreDir(eq(mockNonLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(1, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals(0, storeActions.storesToRestore.size()); }
@Test public void testStoreDeletedWhenCleanDirsFlagSet() { 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); when(mockTaskModel.getTaskMode()).thenReturn(TaskMode.Active); String store1Name = "store1"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); ImmutableMap<SystemStreamPartition, String> mockCheckpointedChangelogOffset = ImmutableMap.of(changelog1SSP, changelog1CheckpointMessage.toString()); Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = new MapConfig(Collections.singletonMap("stores.store1.clean.on.container.start", "true")); Clock mockClock = mock(Clock.class); 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)); }); File dummyCurrentDir = new File("currentDir"); File dummyCheckpointDir = new File("checkpointDir1"); when(mockStorageManagerUtil.getTaskStoreDir(mockLoggedStoreBaseDir, store1Name, taskName, TaskMode.Active)) .thenReturn(dummyCurrentDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(mockLoggedStoreBaseDir, store1Name, taskName, TaskMode.Active)) .thenReturn(ImmutableList.of(dummyCheckpointDir)); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(2, storeActions.storeDirsToDelete.size()); assertTrue(storeActions.storeDirsToDelete.containsValue(dummyCheckpointDir)); assertTrue(storeActions.storeDirsToDelete.containsValue(dummyCurrentDir)); assertEquals("0", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedNonPersistentStore_RestoreToCheckpointedOffset() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(false); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); ImmutableMap<SystemStreamPartition, String> mockCheckpointedChangelogOffset = ImmutableMap.of(changelog1SSP, changelog1CheckpointMessage.toString()); Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(0, storeActions.storeDirsToDelete.size()); assertEquals("0", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedNonPersistentStore_FullRestoreIfCheckpointedOffsetNewerThanNewest() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(false); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "21"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(0, storeActions.storeDirsToDelete.size()); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals("0", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals("10", storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedNonPersistentStore_FullRestoreIfCheckpointedOffsetOlderThanOldest() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(false); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("10", "20", "21"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(0, storeActions.storeDirsToDelete.size()); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals("10", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals("20", storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedNonPersistentStore_FullRestoreIfCheckpointedOffsetInRangeButMaybeCompacted() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(false); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("10", "20", "21"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointId checkpointId = CheckpointId.fromString("0-0"); CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(checkpointId, changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(0, storeActions.storeDirsToDelete.size()); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals("10", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals("20", storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedNonPersistentStore_FullTrimIfNullCheckpointedOffsetAndNotRetainExistingChangelogState() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(false); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = null; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); HashMap<String, String> configMap = new HashMap<>(); configMap.put(TaskConfig.TRANSACTIONAL_STATE_RETAIN_EXISTING_STATE, "false"); Config mockConfig = new MapConfig(configMap); Clock mockClock = mock(Clock.class); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(0, storeActions.storeDirsToDelete.size()); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals("0", storeActions.storesToRestore.get(store1Name).startingOffset); assertNull(storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedNonPersistentStore_FullRestoreIfNullCheckpointedOffsetAndRetainExistingChangelogState() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(false); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = null; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); HashMap<String, String> configMap = new HashMap<>(); configMap.put(TaskConfig.TRANSACTIONAL_STATE_RETAIN_EXISTING_STATE, "true"); Config mockConfig = new MapConfig(configMap); Clock mockClock = mock(Clock.class); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(0, storeActions.storeDirsToDelete.size()); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals("0", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals("10", storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_RestoreToCheckpointedOffsetIfNoStoreCheckpoints() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); ImmutableMap<SystemStreamPartition, String> mockCheckpointedChangelogOffset = ImmutableMap.of(changelog1SSP, changelog1CheckpointMessage.toString()); Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(1, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertEquals("0", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_RestoreToCheckpointedOffsetIfInvalidStoreCheckpoints() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); ImmutableMap<SystemStreamPartition, String> mockCheckpointedChangelogOffset = ImmutableMap.of(changelog1SSP, changelog1CheckpointMessage.toString()); Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreNewerCheckpointDir = mock(File.class); String newerCheckpointDirLocalOffset = "5"; File mockStoreOlderCheckpointDir = mock(File.class); String olderCheckpointDirLocalOffset = "3"; when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(false); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(false); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, newerCheckpointDirLocalOffset)); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, olderCheckpointDirLocalOffset)); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(3, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreOlderCheckpointDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreNewerCheckpointDir)); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals("0", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_RestoreDeltaIfStaleStoreCheckpoint() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); ImmutableMap<SystemStreamPartition, String> mockCheckpointedChangelogOffset = ImmutableMap.of(changelog1SSP, changelog1CheckpointMessage.toString()); Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreNewerCheckpointDir = mock(File.class); String newerCheckpointDirLocalOffset = "4"; File mockStoreOlderCheckpointDir = mock(File.class); String olderCheckpointDirLocalOffset = "3"; when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, newerCheckpointDirLocalOffset)); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, olderCheckpointDirLocalOffset)); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(2, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreOlderCheckpointDir)); assertEquals(mockStoreNewerCheckpointDir, storeActions.storeDirsToRetain.get(store1Name)); assertEquals(newerCheckpointDirLocalOffset, storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_NoRestoreButTrimIfUpToDateStoreCheckpoint() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); ImmutableMap<SystemStreamPartition, String> mockCheckpointedChangelogOffset = ImmutableMap.of(changelog1SSP, changelog1CheckpointMessage.toString()); Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreUpToDateCheckpointDir = mock(File.class); File mockStoreOlderCheckpointDir = mock(File.class); when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreUpToDateCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreUpToDateCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreUpToDateCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, changelog1CheckpointedOffset)); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, "3")); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(2, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreOlderCheckpointDir)); assertEquals(mockStoreUpToDateCheckpointDir, storeActions.storeDirsToRetain.get(store1Name)); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_DeleteStoreCheckpointIfLocalOffsetHigherThanCheckpointed() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); ImmutableMap<SystemStreamPartition, String> mockCheckpointedChangelogOffset = ImmutableMap.of(changelog1SSP, changelog1CheckpointMessage.toString()); Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreNewerCheckpointDir = mock(File.class); File mockStoreOlderCheckpointDir = mock(File.class); when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, "10")); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, changelog1CheckpointedOffset)); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(2, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreNewerCheckpointDir)); assertEquals(mockStoreOlderCheckpointDir, storeActions.storeDirsToRetain.get(store1Name)); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_RetainOneCheckpointIfMultipleCheckpointsWithSameOffset() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); ImmutableMap<SystemStreamPartition, String> mockCheckpointedChangelogOffset = ImmutableMap.of(changelog1SSP, changelog1CheckpointMessage.toString()); Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreNewerCheckpointDir = mock(File.class); File mockStoreOlderCheckpointDir = mock(File.class); when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, changelog1CheckpointedOffset)); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, changelog1CheckpointedOffset)); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(2, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertNotNull(storeActions.storeDirsToRetain.get(store1Name)); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_FullRestoreIfNullCheckpointedOffsetAndRetainExistingChangelogState() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = null; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); HashMap<String, String> configMap = new HashMap<>(); configMap.put(TaskConfig.TRANSACTIONAL_STATE_RETAIN_EXISTING_STATE, "true"); Config mockConfig = new MapConfig(configMap); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreNewerCheckpointDir = mock(File.class); File mockStoreOlderCheckpointDir = mock(File.class); String olderCheckpointDirLocalOffset = "3"; String newerCheckpointDirLocalOffset = "5"; when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, newerCheckpointDirLocalOffset)); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, olderCheckpointDirLocalOffset)); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(3, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreOlderCheckpointDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreNewerCheckpointDir)); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals("0", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals("10", storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_FullTrimIfNullCheckpointedOffsetAndNotRetainExistingState() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = null; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); HashMap<String, String> configMap = new HashMap<>(); configMap.put(TaskConfig.TRANSACTIONAL_STATE_RETAIN_EXISTING_STATE, "false"); Config mockConfig = new MapConfig(configMap); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreNewerCheckpointDir = mock(File.class); File mockStoreOlderCheckpointDir = mock(File.class); String olderCheckpointDirLocalOffset = "3"; String newerCheckpointDirLocalOffset = "5"; when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, newerCheckpointDirLocalOffset)); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, olderCheckpointDirLocalOffset)); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(3, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreOlderCheckpointDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreNewerCheckpointDir)); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals("0", storeActions.storesToRestore.get(store1Name).startingOffset); assertNull(storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_FullRestoreIfEqualCheckpointedOldestAndNewestOffset() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("5", "5", "6"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); HashMap<String, String> configMap = new HashMap<>(); configMap.put(TaskConfig.TRANSACTIONAL_STATE_RETAIN_EXISTING_STATE, "true"); Config mockConfig = new MapConfig(configMap); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreCheckpointDir = mock(File.class); String checkpointDirLocalOffset = "5"; when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, checkpointDirLocalOffset)); Mockito.when(mockSystemAdmin.offsetComparator(anyString(), anyString())) .thenAnswer((Answer<Integer>) invocation -> { String offset1 = (String) invocation.getArguments()[0]; String offset2 = (String) invocation.getArguments()[1]; if (offset1 == null || offset2 == null) { return -1; } return Long.valueOf(offset1).compareTo(Long.valueOf(offset2)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(1, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertEquals(mockStoreCheckpointDir, storeActions.storeDirsToRetain.get(store1Name)); assertEquals("5", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals("5", storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_FullRestoreIfNullCheckpointedAndOldestOffset() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata(null, null, null); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = null; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); HashMap<String, String> configMap = new HashMap<>(); configMap.put(TaskConfig.TRANSACTIONAL_STATE_RETAIN_EXISTING_STATE, "true"); Config mockConfig = new MapConfig(configMap); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreNewerCheckpointDir = mock(File.class); File mockStoreOlderCheckpointDir = mock(File.class); String olderCheckpointDirLocalOffset = "3"; String newerCheckpointDirLocalOffset = "5"; when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, newerCheckpointDirLocalOffset)); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, olderCheckpointDirLocalOffset)); Mockito.when(mockSystemAdmin.offsetComparator(anyString(), anyString())) .thenAnswer((Answer<Integer>) invocation -> { String offset1 = (String) invocation.getArguments()[0]; String offset2 = (String) invocation.getArguments()[1]; if (offset1 == null || offset2 == null) { return -1; } return Long.valueOf(offset1).compareTo(Long.valueOf(offset2)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(3, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreOlderCheckpointDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreNewerCheckpointDir)); assertEquals(0, storeActions.storeDirsToRetain.size()); assertNull(storeActions.storesToRestore.get(store1Name).startingOffset); assertNull(storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_FullRestoreIfCheckpointedOffsetOlderThanOldest() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("11", "20", "21"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreNewerCheckpointDir = mock(File.class); File mockStoreOlderCheckpointDir = mock(File.class); String olderCheckpointDirLocalOffset = "3"; String newerCheckpointDirLocalOffset = "5"; when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, newerCheckpointDirLocalOffset)); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, olderCheckpointDirLocalOffset)); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(3, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreOlderCheckpointDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreNewerCheckpointDir)); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals("11", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals("20", storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_RestoreFromLocalToNewestIfCheckpointedOffsetInRangeButMaybeCompacted() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("4", "20", "21"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "5"; CheckpointId checkpointId = CheckpointId.fromString("0-0"); CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(checkpointId, changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreNewerCheckpointDir = mock(File.class); File mockStoreOlderCheckpointDir = mock(File.class); String olderCheckpointDirLocalOffset = "3"; String newerCheckpointDirLocalOffset = "5"; when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, newerCheckpointDirLocalOffset)); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, olderCheckpointDirLocalOffset)); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(2, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreOlderCheckpointDir)); assertEquals(1, storeActions.storeDirsToRetain.size()); assertEquals(mockStoreNewerCheckpointDir, storeActions.storeDirsToRetain.get(store1Name)); assertEquals("5", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals("20", storeActions.storesToRestore.get(store1Name).endingOffset); }
@Test public void testGetStoreActionsForLoggedPersistentStore_FullRestoreIfCheckpointedOffsetNewerThanNewest() { 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"; StorageEngine store1Engine = mock(StorageEngine.class); StoreProperties mockStore1Properties = mock(StoreProperties.class); when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties); when(mockStore1Properties.isLoggedStore()).thenReturn(true); when(mockStore1Properties.isPersistedToDisk()).thenReturn(true); Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine); String changelog1SystemName = "system1"; String changelog1StreamName = "store1Changelog"; SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName); SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition); SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11"); Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream); String changelog1CheckpointedOffset = "21"; CheckpointedChangelogOffset changelog1CheckpointMessage = new CheckpointedChangelogOffset(CheckpointId.create(), changelog1CheckpointedOffset); Map<SystemStreamPartition, String> mockCheckpointedChangelogOffset = new HashMap<SystemStreamPartition, String>() { { put(changelog1SSP, changelog1CheckpointMessage.toString()); } }; Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets = ImmutableMap.of(changelog1SSP, changelog1SSPMetadata); SystemAdmins mockSystemAdmins = mock(SystemAdmins.class); SystemAdmin mockSystemAdmin = mock(SystemAdmin.class); when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin); StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class); File mockLoggedStoreBaseDir = mock(File.class); File mockNonLoggedStoreBaseDir = mock(File.class); Config mockConfig = mock(Config.class); Clock mockClock = mock(Clock.class); File mockCurrentStoreDir = mock(File.class); File mockStoreNewerCheckpointDir = mock(File.class); File mockStoreOlderCheckpointDir = mock(File.class); String olderCheckpointDirLocalOffset = "5"; String newerCheckpointDirLocalOffset = "15"; when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(mockCurrentStoreDir); when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any())) .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir)); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(), eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true); Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, newerCheckpointDirLocalOffset)); when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false))) .thenReturn(ImmutableMap.of(changelog1SSP, olderCheckpointDirLocalOffset)); 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)); }); StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions( mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil, mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock); assertEquals(3, storeActions.storeDirsToDelete.get(store1Name).size()); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreOlderCheckpointDir)); assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreNewerCheckpointDir)); assertEquals(0, storeActions.storeDirsToRetain.size()); assertEquals("0", storeActions.storesToRestore.get(store1Name).startingOffset); assertEquals("10", storeActions.storesToRestore.get(store1Name).endingOffset); }