@Test public void diskLevelAdaptiveTrackerTest() throws Exception { MockPartitionId mockPartition1 = new MockPartitionId(1L, MockClusterMap.DEFAULT_PARTITION_CLASS); MockPartitionId mockPartition2 = new MockPartitionId(2L, MockClusterMap.DEFAULT_PARTITION_CLASS); List<Port> portList = Collections.singletonList(new Port(PORT, PortType.PLAINTEXT)); List<String> mountPaths = Arrays.asList("mockMountPath0", "mockMountPath1"); MockDataNodeId localHost1 = new MockDataNodeId("LocalHost1", portList, mountPaths, "dc-0"); MockDataNodeId remoteHost1 = new MockDataNodeId("RemoteHost1", portList, mountPaths, "dc-1"); List<MockDataNodeId> datanodes = new ArrayList<>(Arrays.asList(localHost1, remoteHost1)); ReplicaId partition1Replica1 = new MockReplicaId(PORT, mockPartition1, localHost1, 0); ReplicaId partition1Replica2 = new MockReplicaId(PORT, mockPartition1, remoteHost1, 0); ReplicaId partition1Replica3 = new MockReplicaId(PORT, mockPartition1, remoteHost1, 1); ReplicaId partition2Replica1 = new MockReplicaId(PORT, mockPartition2, localHost1, 1); mockPartition1.replicaIds.add(partition1Replica1); mockPartition1.replicaIds.add(partition1Replica2); mockPartition1.replicaIds.add(partition1Replica3); mockPartition2.replicaIds.add(partition2Replica1); mockPartition2.replicaIds.add(new MockReplicaId(PORT, mockPartition2, remoteHost1, 0)); mockPartition2.replicaIds.add(new MockReplicaId(PORT, mockPartition2, remoteHost1, 1)); MockClusterMap clusterMap = new MockClusterMap(false, datanodes, 2, Arrays.asList(mockPartition1, mockPartition2), localDcName); trackerScope = OperationTrackerScope.Disk; RouterConfig routerConfig = createRouterConfig(true, 1, 1, 6, null, true); NonBlockingRouterMetrics originalMetrics = routerMetrics; routerMetrics = new NonBlockingRouterMetrics(clusterMap, routerConfig); Counter pastDueCount = routerMetrics.getBlobPastDueCount; Map<Resource, CachedHistogram> localColoMap = routerMetrics.getBlobLocalDcResourceToLatency; Map<Resource, CachedHistogram> crossColoMap = routerMetrics.getBlobCrossDcResourceToLatency; Histogram localHostDisk0Histogram = localColoMap.get(partition1Replica1.getDiskId()); Histogram localHostDisk1Histogram = localColoMap.get(partition2Replica1.getDiskId()); Histogram remoteHostDisk0Histogram = crossColoMap.get(partition1Replica2.getDiskId()); Histogram remoteHostDisk1Histogram = crossColoMap.get(partition1Replica3.getDiskId()); primeTracker(localHostDisk0Histogram, routerConfig.routerOperationTrackerMinDataPointsRequired, new Pair<>(0L, 50L)); primeTracker(localHostDisk1Histogram, routerConfig.routerOperationTrackerMinDataPointsRequired, new Pair<>(100L, 120L)); primeTracker(remoteHostDisk0Histogram, routerConfig.routerOperationTrackerMinDataPointsRequired, new Pair<>(150L, 180L)); primeTracker(remoteHostDisk1Histogram, routerConfig.routerOperationTrackerMinDataPointsRequired, new Pair<>(150L, 180L)); double localHostDisk0Cutoff = localHostDisk0Histogram.getSnapshot().getValue(QUANTILE); double localHostDisk1Cutoff = localHostDisk1Histogram.getSnapshot().getValue(QUANTILE); double remoteHostDisk0Cutoff = remoteHostDisk0Histogram.getSnapshot().getValue(QUANTILE); OperationTracker tracker1 = getOperationTracker(routerConfig, mockPartition1); OperationTracker tracker2 = getOperationTracker(routerConfig, mockPartition2); sendRequests(tracker2, 1); sendRequests(tracker1, 1); time.sleep((long) localHostDisk0Cutoff + 1); sendRequests(tracker1, 1); sendRequests(tracker2, 0); time.sleep((long) (localHostDisk1Cutoff - localHostDisk0Cutoff) + 2); sendRequests(tracker1, 0); sendRequests(tracker2, 1); time.sleep((long) remoteHostDisk0Cutoff + 1); sendRequests(tracker1, 1); sendRequests(tracker2, 1); assertFalse("Operation should not be done", tracker1.isDone() || tracker2.isDone()); tracker1.onResponse(partitionAndInflightReplicas.get(mockPartition1).poll(), TrackedRequestFinalState.SUCCESS); tracker2.onResponse(partitionAndInflightReplicas.get(mockPartition2).poll(), TrackedRequestFinalState.SUCCESS); tracker1.onResponse(partitionAndInflightReplicas.get(mockPartition1).poll(), TrackedRequestFinalState.FAILURE); tracker2.onResponse(partitionAndInflightReplicas.get(mockPartition2).poll(), TrackedRequestFinalState.TIMED_OUT); tracker1.onResponse(partitionAndInflightReplicas.get(mockPartition1).poll(), TrackedRequestFinalState.TIMED_OUT); tracker2.onResponse(partitionAndInflightReplicas.get(mockPartition2).poll(), TrackedRequestFinalState.FAILURE); assertTrue("Operation should have succeeded", tracker1.hasSucceeded() && tracker2.hasSucceeded()); assertEquals("Past due counter is not expected", 4, pastDueCount.getCount()); assertEquals("Mismatch in number of data points in local colo histogram", 2, routerMetrics.getBlobLocalDcLatencyMs.getCount()); assertEquals("Mismatch in number of data points in cross colo histogram", 2, routerMetrics.getBlobCrossDcLatencyMs.getCount()); MockDataNodeId newNode1 = clusterMap.createNewDataNodes(1, "dc-0").get(0); MockDataNodeId newNode2 = clusterMap.createNewDataNodes(1, "dc-1").get(0); MockPartitionId mockPartition3 = new MockPartitionId(3L, MockClusterMap.DEFAULT_PARTITION_CLASS); mockPartition3.replicaIds.add(new MockReplicaId(PORT, mockPartition3, newNode1, 0)); mockPartition3.replicaIds.add(new MockReplicaId(PORT, mockPartition3, newNode2, 1)); OperationTracker tracker3 = getOperationTracker(routerConfig, mockPartition3); sendRequests(tracker3, 1); sendRequests(tracker3, 0); tracker3.onResponse(partitionAndInflightReplicas.get(mockPartition3).poll(), TrackedRequestFinalState.FAILURE); sendRequests(tracker3, 1); tracker3.onResponse(partitionAndInflightReplicas.get(mockPartition3).poll(), TrackedRequestFinalState.SUCCESS); assertTrue("Operation should have succeeded", tracker3.hasSucceeded()); trackerScope = OperationTrackerScope.Datacenter; routerMetrics = originalMetrics; }
@Override public void onResponse(ReplicaId replicaId, TrackedRequestFinalState trackedRequestFinalState) { super.onResponse(replicaId, trackedRequestFinalState); long elapsedTime; if (unexpiredRequestSendTimes.containsKey(replicaId)) { elapsedTime = time.milliseconds() - unexpiredRequestSendTimes.remove(replicaId).getSecond(); } else { elapsedTime = time.milliseconds() - expiredRequestSendTimes.remove(replicaId); } if (trackedRequestFinalState != TrackedRequestFinalState.TIMED_OUT || !routerConfig.routerOperationTrackerExcludeTimeoutEnabled) { getLatencyHistogram(replicaId).update(elapsedTime); if (routerConfig.routerOperationTrackerMetricScope != OperationTrackerScope.Datacenter) { Histogram histogram = replicaId.getDataNodeId().getDatacenterName().equals(datacenterName) ? localDcHistogram : crossDcHistogram; histogram.update(elapsedTime); } } }
AdaptiveOperationTracker extends SimpleOperationTracker { @Override public void onResponse(ReplicaId replicaId, TrackedRequestFinalState trackedRequestFinalState) { super.onResponse(replicaId, trackedRequestFinalState); long elapsedTime; if (unexpiredRequestSendTimes.containsKey(replicaId)) { elapsedTime = time.milliseconds() - unexpiredRequestSendTimes.remove(replicaId).getSecond(); } else { elapsedTime = time.milliseconds() - expiredRequestSendTimes.remove(replicaId); } if (trackedRequestFinalState != TrackedRequestFinalState.TIMED_OUT || !routerConfig.routerOperationTrackerExcludeTimeoutEnabled) { getLatencyHistogram(replicaId).update(elapsedTime); if (routerConfig.routerOperationTrackerMetricScope != OperationTrackerScope.Datacenter) { Histogram histogram = replicaId.getDataNodeId().getDatacenterName().equals(datacenterName) ? localDcHistogram : crossDcHistogram; histogram.update(elapsedTime); } } } }
AdaptiveOperationTracker extends SimpleOperationTracker { @Override public void onResponse(ReplicaId replicaId, TrackedRequestFinalState trackedRequestFinalState) { super.onResponse(replicaId, trackedRequestFinalState); long elapsedTime; if (unexpiredRequestSendTimes.containsKey(replicaId)) { elapsedTime = time.milliseconds() - unexpiredRequestSendTimes.remove(replicaId).getSecond(); } else { elapsedTime = time.milliseconds() - expiredRequestSendTimes.remove(replicaId); } if (trackedRequestFinalState != TrackedRequestFinalState.TIMED_OUT || !routerConfig.routerOperationTrackerExcludeTimeoutEnabled) { getLatencyHistogram(replicaId).update(elapsedTime); if (routerConfig.routerOperationTrackerMetricScope != OperationTrackerScope.Datacenter) { Histogram histogram = replicaId.getDataNodeId().getDatacenterName().equals(datacenterName) ? localDcHistogram : crossDcHistogram; histogram.update(elapsedTime); } } } AdaptiveOperationTracker(RouterConfig routerConfig, NonBlockingRouterMetrics routerMetrics,
      RouterOperation routerOperation, PartitionId partitionId, String originatingDcName, Time time); }
AdaptiveOperationTracker extends SimpleOperationTracker { @Override public void onResponse(ReplicaId replicaId, TrackedRequestFinalState trackedRequestFinalState) { super.onResponse(replicaId, trackedRequestFinalState); long elapsedTime; if (unexpiredRequestSendTimes.containsKey(replicaId)) { elapsedTime = time.milliseconds() - unexpiredRequestSendTimes.remove(replicaId).getSecond(); } else { elapsedTime = time.milliseconds() - expiredRequestSendTimes.remove(replicaId); } if (trackedRequestFinalState != TrackedRequestFinalState.TIMED_OUT || !routerConfig.routerOperationTrackerExcludeTimeoutEnabled) { getLatencyHistogram(replicaId).update(elapsedTime); if (routerConfig.routerOperationTrackerMetricScope != OperationTrackerScope.Datacenter) { Histogram histogram = replicaId.getDataNodeId().getDatacenterName().equals(datacenterName) ? localDcHistogram : crossDcHistogram; histogram.update(elapsedTime); } } } AdaptiveOperationTracker(RouterConfig routerConfig, NonBlockingRouterMetrics routerMetrics,
      RouterOperation routerOperation, PartitionId partitionId, String originatingDcName, Time time); @Override void onResponse(ReplicaId replicaId, TrackedRequestFinalState trackedRequestFinalState); @Override Iterator<ReplicaId> getReplicaIterator(); }
AdaptiveOperationTracker extends SimpleOperationTracker { @Override public void onResponse(ReplicaId replicaId, TrackedRequestFinalState trackedRequestFinalState) { super.onResponse(replicaId, trackedRequestFinalState); long elapsedTime; if (unexpiredRequestSendTimes.containsKey(replicaId)) { elapsedTime = time.milliseconds() - unexpiredRequestSendTimes.remove(replicaId).getSecond(); } else { elapsedTime = time.milliseconds() - expiredRequestSendTimes.remove(replicaId); } if (trackedRequestFinalState != TrackedRequestFinalState.TIMED_OUT || !routerConfig.routerOperationTrackerExcludeTimeoutEnabled) { getLatencyHistogram(replicaId).update(elapsedTime); if (routerConfig.routerOperationTrackerMetricScope != OperationTrackerScope.Datacenter) { Histogram histogram = replicaId.getDataNodeId().getDatacenterName().equals(datacenterName) ? localDcHistogram : crossDcHistogram; histogram.update(elapsedTime); } } } AdaptiveOperationTracker(RouterConfig routerConfig, NonBlockingRouterMetrics routerMetrics,
      RouterOperation routerOperation, PartitionId partitionId, String originatingDcName, Time time); @Override void onResponse(ReplicaId replicaId, TrackedRequestFinalState trackedRequestFinalState); @Override Iterator<ReplicaId> getReplicaIterator();  }