@Test public void partitionLevelAdaptiveTrackerTest() throws Exception { MockPartitionId mockPartition1 = new MockPartitionId(0L, MockClusterMap.DEFAULT_PARTITION_CLASS); MockPartitionId mockPartition2 = new MockPartitionId(1L, MockClusterMap.DEFAULT_PARTITION_CLASS); for (int i = 0; i < REPLICA_COUNT; i++) { mockPartition1.replicaIds.add(new MockReplicaId(PORT, mockPartition1, datanodes.get(i % datanodes.size()), 1)); mockPartition2.replicaIds.add(new MockReplicaId(PORT, mockPartition2, datanodes.get(i % datanodes.size()), 2)); } MockClusterMap clusterMap = new MockClusterMap(false, datanodes, 3, Arrays.asList(mockPartition1, mockPartition2), localDcName); trackerScope = OperationTrackerScope.Partition; RouterConfig routerConfig = createRouterConfig(true, 2, 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 localHistogram1 = localColoMap.get(mockPartition1); Histogram localHistogram2 = localColoMap.get(mockPartition2); Histogram remoteHistogram1 = crossColoMap.get(mockPartition1); primeTracker(localHistogram1, routerConfig.routerOperationTrackerMinDataPointsRequired, new Pair<>(0L, 50L)); primeTracker(localHistogram2, routerConfig.routerOperationTrackerMinDataPointsRequired, new Pair<>(100L, 120L)); primeTracker(remoteHistogram1, routerConfig.routerOperationTrackerMinDataPointsRequired, new Pair<>(150L, 180L)); OperationTracker tracker1 = getOperationTracker(routerConfig, mockPartition1); OperationTracker tracker2 = getOperationTracker(routerConfig, mockPartition2); double localColoCutoff1 = localHistogram1.getSnapshot().getValue(QUANTILE); double localColoCutoff2 = localHistogram2.getSnapshot().getValue(QUANTILE); double crossColoCutoff1 = remoteHistogram1.getSnapshot().getValue(QUANTILE); sendRequests(tracker2, 1); sendRequests(tracker1, 1); time.sleep((long) localColoCutoff1 + 1); sendRequests(tracker1, 1); sendRequests(tracker2, 0); time.sleep((long) (localColoCutoff2 - localColoCutoff1) + 2); sendRequests(tracker1, 1); sendRequests(tracker2, 1); time.sleep((long) localColoCutoff1 + 1); sendRequests(tracker1, 1); sendRequests(tracker2, 0); time.sleep((long) crossColoCutoff1 + 1); sendRequests(tracker1, 1); sendRequests(tracker2, 1); for (int i = 0; i < 2; ++i) { 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); } assertTrue("Operation should have succeeded", tracker1.hasSucceeded() && tracker2.hasSucceeded()); assertEquals("Past due counter is not expected", 4 + 2, pastDueCount.getCount()); LinkedList<ReplicaId> inflightRequests1 = partitionAndInflightReplicas.get(mockPartition1); LinkedList<ReplicaId> inflightRequests2 = partitionAndInflightReplicas.get(mockPartition2); while (!inflightRequests1.isEmpty()) { tracker1.onResponse(inflightRequests1.poll(), TrackedRequestFinalState.FAILURE); } while (!inflightRequests2.isEmpty()) { tracker2.onResponse(inflightRequests2.poll(), TrackedRequestFinalState.TIMED_OUT); } assertEquals("Mismatch in number of data points in local colo histogram", 5, routerMetrics.getBlobLocalDcLatencyMs.getCount()); assertEquals("Mismatch in number of data points in cross colo histogram", 2, routerMetrics.getBlobCrossDcLatencyMs.getCount()); MockPartitionId mockPartition3 = (MockPartitionId) clusterMap.createNewPartition(datanodes); OperationTracker tracker3 = getOperationTracker(routerConfig, mockPartition3); sendRequests(tracker3, 1); sendRequests(tracker3, 0); tracker3.onResponse(partitionAndInflightReplicas.get(mockPartition3).poll(), TrackedRequestFinalState.SUCCESS); 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();  }