@Test
  public void testPutGetDeleteSequence() throws Exception {
    prepareDb(1000);

    List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS);
    List<Long> partitions = partitionIds.stream().map(p -> p.getId()).collect(Collectors.toList());
    int blobsPerContainer = 5;

    String hostName1 = "localhost1";
    String hostName2 = "localhost2";
    int hostPort1 = 6024;
    int hostPort2 = 6025;
    // simulate the requests which will be sent on this host (thisNodeName, thisNodePort)
    String thisNodeName = hostName1;
    int thisNodePort = hostPort1;

    // Prepare RepairRequests and insert them to the DB.
    // Map<Partition ID, Map<BlobId, RepairRequestRecord>>
    Map<Integer, Map<String, RepairRequestRecord>> records = new HashMap<>();
    for (Account account : accountService.getAllAccounts()) {
      for (Container container : account.getAllContainers()) {
        for (int i = 0; i < blobsPerContainer; i++) {
          PartitionId partitionId = partitionIds.get(random.nextInt(partitionIds.size()));
          String blobId = generateBlobId(account, container, partitionId);
          RepairRequestRecord.OperationType operationType = i % 2 == 0 ? TtlUpdateRequest : DeleteRequest;
          long operationTime = System.currentTimeMillis() - random.nextInt(1000);
          short lifeVersion = -1;
          long expirationTime =
              i % 2 == 0 ? Utils.Infinite_Time : System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1);
          String hostName = random.nextInt(2) == 0 ? hostName1 : hostName2;
          int hostPort = random.nextInt(2) == 0 ? hostPort1 : hostPort2;
          RepairRequestRecord record =
              new RepairRequestRecord(blobId, (int) partitionId.getId(), hostName, hostPort, operationType,
                  operationTime, lifeVersion, expirationTime);
          repairRequestsDb.putRepairRequests(record);
          records.putIfAbsent((int) partitionId.getId(), new HashMap<>());
          records.get((int) partitionId.getId()).put(record.getBlobId(), record);
        }
      }
    }

    // on one node with name as thisNodeName and port as thisNodePort,
    // read the database but exclude all the records which has the source replica as this node.
    // we should run ODR to fix the requests on the nodes except the source replica
    Set<Long> partitionsNeedRepair = repairRequestsDb.getPartitionsNeedRepair(thisNodeName, thisNodePort, partitions);
    for (PartitionId id : partitionIds) {
      Pair<List<RepairRequestRecord>, Long> dbRecords =
          repairRequestsDb.getRepairRequestsExcludingHost((int) id.getId(), thisNodeName, thisNodePort, 0);
      List<RepairRequestRecord> recordFromStore = dbRecords.getFirst();
      Set<Long> partitionsNeedRepairUpdated;
      Map<String, RepairRequestRecord> orgRecords = records.get((int) id.getId());
      if (recordFromStore.size() > 0) {
        for (RepairRequestRecord record : recordFromStore) {
          RepairRequestRecord org = orgRecords.get(record.getBlobId());
          assertEquals("Record does not match expectation ", org, record);
          assertTrue("should exclude this node",
              !thisNodeName.equals(record.getSourceHostName()) || thisNodePort != record.getSourceHostPort());
          orgRecords.remove(record.getBlobId());
          repairRequestsDb.removeRepairRequests(record.getBlobId(), record.getOperationType());
        }
        partitionsNeedRepairUpdated = repairRequestsDb.getPartitionsNeedRepair(thisNodeName, thisNodePort, partitions);
        assertTrue(partitionsNeedRepair.contains(id.getId()));
        assertFalse(partitionsNeedRepairUpdated.contains(id.getId()));
        assertEquals(partitionsNeedRepair.size(), partitionsNeedRepairUpdated.size() + 1);
      } else {
        partitionsNeedRepairUpdated = repairRequestsDb.getPartitionsNeedRepair(thisNodeName, thisNodePort, partitions);
        assertEquals(partitionsNeedRepairUpdated, partitionsNeedRepair);
        assertFalse(partitionsNeedRepairUpdated.contains(id.getId()));
      }
      partitionsNeedRepair = partitionsNeedRepairUpdated;
    }
    assertTrue(partitionsNeedRepair.size() == 0);

    // get the remaining records.
    for (PartitionId id : partitionIds) {
      List<RepairRequestRecord> recordFromStore = repairRequestsDb.getRepairRequestsForPartition((int) id.getId());
      Map<String, RepairRequestRecord> orgRecords = records.get((int) id.getId());
      assertEquals("Record number doesn't match.", orgRecords.size(), recordFromStore.size());
      for (RepairRequestRecord record : recordFromStore) {
        RepairRequestRecord org = orgRecords.get(record.getBlobId());
        assertEquals("Record does not match expectation ", org, record);
        assertEquals(thisNodeName, record.getSourceHostName());
        assertEquals(thisNodePort, record.getSourceHostPort());
        orgRecords.remove(record.getBlobId());
        repairRequestsDb.removeRepairRequests(record.getBlobId(), record.getOperationType());
      }
      assertTrue("Should be emptry now.", orgRecords.isEmpty());
      records.remove((int) id.getId());
    }
    assertTrue("Should have zero records now", records.isEmpty());

    // delete the records and check that they cannot be fetched with a get call.
    for (PartitionId id : partitionIds) {
      List<RepairRequestRecord> recordFromStore = repairRequestsDb.getRepairRequestsForPartition((int) id.getId());
      assertTrue("No more records", recordFromStore.isEmpty());
    }

    records.clear();
  }
Tests sequences of puts, gets and deletes across multiple containers.
@throws Exception
ambry-mysql/src/integration-test/java/com/github/ambry/repair/MysqlRepairRequestsDbTest.java