@Test
  public void blockDeprecatedContainerReplicationTest() throws Exception {
    Properties properties = new Properties();
    properties.setProperty("replication.container.deletion.enabled", "true");
    replicationConfig = new ReplicationConfig(new VerifiableProperties(properties));
    MockClusterMap clusterMap = new MockClusterMap();
    Pair<MockHost, MockHost> localAndRemoteHosts = getLocalAndRemoteHosts(clusterMap);
    MockHost localHost = localAndRemoteHosts.getFirst();
    MockHost remoteHost = localAndRemoteHosts.getSecond();
    MockStoreKeyConverterFactory storeKeyConverterFactory = new MockStoreKeyConverterFactory(null, null);
    storeKeyConverterFactory.setConversionMap(new HashMap<>());
    storeKeyConverterFactory.setReturnInputIfAbsent(true);
    MockStoreKeyConverterFactory.MockStoreKeyConverter storeKeyConverter =
        storeKeyConverterFactory.getStoreKeyConverter();
    Map<StoreKey, StoreKey> conversionMap = new HashMap<>();
    storeKeyConverter.setConversionMap(conversionMap);
    List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(null);

    for (int i = 0; i < partitionIds.size(); i++) {
      PartitionId partitionId = partitionIds.get(i);
      BlobId b0 = generateRandomBlobId(partitionId);
      conversionMap.put(b0, b0);
      BlobId b1 = generateRandomBlobId(partitionId);
      conversionMap.put(b1, b1);
      // add 2 messages to both hosts.
      storeKeyConverter.setConversionMap(conversionMap);
      storeKeyConverter.convert(conversionMap.keySet());
      //addPutMessagesToReplicasOfPartition(Arrays.asList(b0), Arrays.asList(localHost, remoteHost));
      // add 3 messages to the remote host only
      addPutMessagesToReplicasOfPartition(Arrays.asList(b0, b1), Collections.singletonList(remoteHost));
    }

    StoreKeyFactory storeKeyFactory = new BlobIdFactory(clusterMap);
    Transformer transformer = new BlobIdTransformer(storeKeyFactory, storeKeyConverter);
    int batchSize = 4;
    ReplicationMetrics replicationMetrics =
        new ReplicationMetrics(new MetricRegistry(), clusterMap.getReplicaIds(localHost.dataNodeId));
    replicationMetrics.populateSingleColoMetrics(remoteHost.dataNodeId.getDatacenterName());

    List<RemoteReplicaInfo> remoteReplicaInfoList = localHost.getRemoteReplicaInfos(remoteHost, null);
    Map<DataNodeId, List<RemoteReplicaInfo>> replicasToReplicate =
        Collections.singletonMap(remoteHost.dataNodeId, remoteReplicaInfoList);
    storeKeyFactory = Utils.getObj("com.github.ambry.commons.BlobIdFactory", clusterMap);
    Map<DataNodeId, MockHost> hosts = new HashMap<>();
    hosts.put(remoteHost.dataNodeId, remoteHost);
    MockConnectionPool connectionPool = new MockConnectionPool(hosts, clusterMap, batchSize);
    Predicate<MessageInfo> skipPredicate = new ReplicationSkipPredicate(accountService, replicationConfig);
    MockFindTokenHelper findTokenHelper = new MockFindTokenHelper(storeKeyFactory, replicationConfig);
    MockNetworkClient networkClient = new MockNetworkClient(hosts, clusterMap, 4, findTokenHelper);
    ReplicaThread replicaThread =
        new ReplicaThread("threadtest", findTokenHelper, clusterMap, new AtomicInteger(0), localHost.dataNodeId,
            networkClient, replicationConfig, replicationMetrics, null, storeKeyConverter, transformer,
            clusterMap.getMetricRegistry(), false, localHost.dataNodeId.getDatacenterName(),
            new ResponseHandler(clusterMap), time, null, skipPredicate, null);
    for (RemoteReplicaInfo remoteReplicaInfo : remoteReplicaInfoList) {
      replicaThread.addRemoteReplicaInfo(remoteReplicaInfo);
    }
    List<RemoteReplicaInfo> remoteReplicaInfos = replicasToReplicate.get(remoteHost.dataNodeId);

    DataNodeId remoteNode = remoteReplicaInfos.get(0).getReplicaId().getDataNodeId();
    ReplicaMetadataResponse response = replicaThread.getReplicaMetadataResponse(remoteReplicaInfos,
        new MockConnectionPool.MockConnection(remoteHost, batchSize), remoteNode);
    //case1 DELETE_IN_PROGRESS container with retention time qualified.
    for (int i = 0; i < 2; i++) {
      RemoteReplicaInfo remoteReplicaInfo = remoteReplicaInfos.get(i);
      ReplicaMetadataResponseInfo replicaMetadataResponseInfo = response.getReplicaMetadataResponseInfoList().get(i);
      new ResponseHandler(clusterMap).onEvent(remoteReplicaInfo.getReplicaId(), replicaMetadataResponseInfo.getError());
      for (int j = 0; j < replicaMetadataResponseInfo.getMessageInfoList().size(); j++) {
        short accountId = replicaMetadataResponseInfo.getMessageInfoList().get(j).getAccountId();
        short containerId = replicaMetadataResponseInfo.getMessageInfoList().get(j).getContainerId();
        Container container = Mockito.mock(Container.class);
        Account account = Mockito.mock(Account.class);
        Mockito.when(account.getContainerById(containerId)).thenReturn(container);
        Mockito.when(accountService.getAccountById(accountId)).thenReturn(account);
        Mockito.when(container.getDeleteTriggerTime())
            .thenReturn(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(
                replicationConfig.replicationContainerDeletionRetentionDays + 1));
        Mockito.when(container.getStatus()).thenReturn(Container.ContainerStatus.DELETE_IN_PROGRESS);
      }
      Set<MessageInfo> remoteMissingStoreKeys =
          replicaThread.getMissingStoreMessages(replicaMetadataResponseInfo, remoteNode, remoteReplicaInfo);
      assertEquals("All DELETE_IN_PROGRESS blobs qualified with retention time should be skipped during replication", 0,
          remoteMissingStoreKeys.size());
      Map<StoreKey, StoreKey> remoteKeyToLocalKeyMap = replicaThread.batchConvertReplicaMetadataResponseKeys(response, new ArrayList<>());
      replicaThread.processReplicaMetadataResponse(remoteMissingStoreKeys, replicaMetadataResponseInfo,
          remoteReplicaInfo, remoteNode, remoteKeyToLocalKeyMap);
    }
    //case2 DELETE_IN_PROGRESS container with retention time not qualified.
    for (int i = 2; i < 4; i++) {
      RemoteReplicaInfo remoteReplicaInfo = remoteReplicaInfos.get(i);
      ReplicaMetadataResponseInfo replicaMetadataResponseInfo = response.getReplicaMetadataResponseInfoList().get(i);
      new ResponseHandler(clusterMap).onEvent(remoteReplicaInfo.getReplicaId(), replicaMetadataResponseInfo.getError());
      for (int j = 0; j < replicaMetadataResponseInfo.getMessageInfoList().size(); j++) {
        short accountId = replicaMetadataResponseInfo.getMessageInfoList().get(j).getAccountId();
        short containerId = replicaMetadataResponseInfo.getMessageInfoList().get(j).getContainerId();
        Container container = Mockito.mock(Container.class);
        Account account = Mockito.mock(Account.class);
        Mockito.when(account.getContainerById(containerId)).thenReturn(container);
        Mockito.when(accountService.getAccountById(accountId)).thenReturn(account);
        Mockito.when(container.getStatus()).thenReturn(Container.ContainerStatus.DELETE_IN_PROGRESS);
        Mockito.when(container.getDeleteTriggerTime()).thenReturn(System.currentTimeMillis());
      }
      Set<MessageInfo> remoteMissingStoreKeys =
          replicaThread.getMissingStoreMessages(replicaMetadataResponseInfo, remoteNode, remoteReplicaInfo);
      assertEquals(
          "All DELETE_IN_PROGRESS blobs not qualified with retention time should not be skipped during replication", 2,
          remoteMissingStoreKeys.size());
      Map<StoreKey, StoreKey> remoteKeyToLocalKeyMap = replicaThread.batchConvertReplicaMetadataResponseKeys(response, new ArrayList<>());
      replicaThread.processReplicaMetadataResponse(remoteMissingStoreKeys, replicaMetadataResponseInfo,
          remoteReplicaInfo, remoteNode, remoteKeyToLocalKeyMap);
    }
    //case3 INACTIVE container
    for (int i = 4; i < 6; i++) {
      RemoteReplicaInfo remoteReplicaInfo = remoteReplicaInfos.get(i);
      ReplicaMetadataResponseInfo replicaMetadataResponseInfo = response.getReplicaMetadataResponseInfoList().get(i);
      new ResponseHandler(clusterMap).onEvent(remoteReplicaInfo.getReplicaId(), replicaMetadataResponseInfo.getError());
      for (int j = 0; j < replicaMetadataResponseInfo.getMessageInfoList().size(); j++) {
        short accountId = replicaMetadataResponseInfo.getMessageInfoList().get(j).getAccountId();
        short containerId = replicaMetadataResponseInfo.getMessageInfoList().get(j).getContainerId();
        Container container = Mockito.mock(Container.class);
        Account account = Mockito.mock(Account.class);
        Mockito.when(account.getContainerById(containerId)).thenReturn(container);
        Mockito.when(accountService.getAccountById(accountId)).thenReturn(account);
        Mockito.when(container.getStatus()).thenReturn(Container.ContainerStatus.INACTIVE);
      }
      Set<MessageInfo> remoteMissingStoreKeys =
          replicaThread.getMissingStoreMessages(replicaMetadataResponseInfo, remoteNode, remoteReplicaInfo);
      assertEquals("All INACTIVE blobs should be skipped during replication", 0, remoteMissingStoreKeys.size());
      Map<StoreKey, StoreKey> remoteKeyToLocalKeyMap = replicaThread.batchConvertReplicaMetadataResponseKeys(response, new ArrayList<>());
      replicaThread.processReplicaMetadataResponse(remoteMissingStoreKeys, replicaMetadataResponseInfo,
          remoteReplicaInfo, remoteNode, remoteKeyToLocalKeyMap);
    }
    //case 4 ACTIVE Container
    for (int i = 6; i < 8; i++) {
      RemoteReplicaInfo remoteReplicaInfo = remoteReplicaInfos.get(i);
      ReplicaMetadataResponseInfo replicaMetadataResponseInfo = response.getReplicaMetadataResponseInfoList().get(i);
      new ResponseHandler(clusterMap).onEvent(remoteReplicaInfo.getReplicaId(), replicaMetadataResponseInfo.getError());
      for (int j = 0; j < replicaMetadataResponseInfo.getMessageInfoList().size(); j++) {
        short accountId = replicaMetadataResponseInfo.getMessageInfoList().get(j).getAccountId();
        short containerId = replicaMetadataResponseInfo.getMessageInfoList().get(j).getContainerId();
        Container container = Mockito.mock(Container.class);
        Account account = Mockito.mock(Account.class);
        Mockito.when(account.getContainerById(containerId)).thenReturn(container);
        Mockito.when(accountService.getAccountById(accountId)).thenReturn(account);
        Mockito.when(container.getStatus()).thenReturn(Container.ContainerStatus.ACTIVE);
      }
      Set<MessageInfo> remoteMissingStoreKeys =
          replicaThread.getMissingStoreMessages(replicaMetadataResponseInfo, remoteNode, remoteReplicaInfo);
      assertEquals("All non-deprecated blobs should not be skipped during replication", 2,
          remoteMissingStoreKeys.size());
      Map<StoreKey, StoreKey> remoteKeyToLocalKeyMap = replicaThread.batchConvertReplicaMetadataResponseKeys(response, new ArrayList<>());
      replicaThread.processReplicaMetadataResponse(remoteMissingStoreKeys, replicaMetadataResponseInfo,
          remoteReplicaInfo, remoteNode, remoteKeyToLocalKeyMap);
    }
  }
Tests if deprecated containers have been blocked during replication.
ambry-replication/src/test/java/com/github/ambry/replication/ReplicationTest.java