@Test public void testDropDataMetricTriggersAutoScale() throws InterruptedException { final String jobId = "test-job-1"; final int stage = 1; final int workerIdx = 0; final int workerNum = 1; final int dropCount = 1; final int onNextCount = 9; final double dropPercent = dropCount * 100.0 / (dropCount + onNextCount); final List<GaugeMeasurement> gauges = Arrays.asList( new GaugeMeasurement(MetricStringConstants.ON_NEXT_COUNT, onNextCount), new GaugeMeasurement(MetricStringConstants.DROP_COUNT, dropCount)); final MantisMasterClientApi mockMasterClientApi = mock(MantisMasterClientApi.class); final Map<Integer, WorkerAssignments> assignmentsMap = new HashMap<>(); assignmentsMap.put(stage, new WorkerAssignments(stage, 1, Collections.singletonMap(1, new WorkerHost("localhost", workerIdx, Arrays.asList(31300), MantisJobState.Started, workerNum, 31301, -1)))); when(mockMasterClientApi.schedulingChanges(jobId)).thenReturn(Observable.just(new JobSchedulingInfo(jobId, assignmentsMap))); final CountDownLatch latch = new CountDownLatch(1); final AutoScaleMetricsConfig aggregationConfig = new AutoScaleMetricsConfig(); final WorkerMetricHandler workerMetricHandler = new WorkerMetricHandler(jobId, new Observer<JobAutoScaler.Event>() { @Override public void onCompleted() { logger.warn("onCompleted"); } @Override public void onError(Throwable e) { logger.warn("onError {}", e.getMessage(), e); } @Override public void onNext(JobAutoScaler.Event event) { logger.info("got auto scale event {}", event); JobAutoScaler.Event expected = new JobAutoScaler.Event(StageScalingPolicy.ScalingReason.DataDrop, stage, dropPercent, 1, ""); assertEquals(expected, event); latch.countDown(); } }, mockMasterClientApi, aggregationConfig); final Observer<MetricData> metricDataObserver = workerMetricHandler.initAndGetMetricDataObserver(); metricDataObserver.onNext(new MetricData(jobId, stage, workerIdx, workerNum, DATA_DROP_METRIC_GROUP, gauges)); assertTrue(latch.await(30 + 5, TimeUnit.SECONDS)); }
public Observer<MetricData> initAndGetMetricDataObserver() { start(); return new SerializedObserver<>(metricDataSubject); }
WorkerMetricHandler { public Observer<MetricData> initAndGetMetricDataObserver() { start(); return new SerializedObserver<>(metricDataSubject); } }
WorkerMetricHandler { public Observer<MetricData> initAndGetMetricDataObserver() { start(); return new SerializedObserver<>(metricDataSubject); } WorkerMetricHandler(final String jobId,
                               final Observer<JobAutoScaler.Event> jobAutoScaleObserver,
                               final MantisMasterClientApi masterClientApi,
                               final AutoScaleMetricsConfig autoScaleMetricsConfig); }
WorkerMetricHandler { public Observer<MetricData> initAndGetMetricDataObserver() { start(); return new SerializedObserver<>(metricDataSubject); } WorkerMetricHandler(final String jobId,
                               final Observer<JobAutoScaler.Event> jobAutoScaleObserver,
                               final MantisMasterClientApi masterClientApi,
                               final AutoScaleMetricsConfig autoScaleMetricsConfig); Observer<MetricData> initAndGetMetricDataObserver(); }
WorkerMetricHandler { public Observer<MetricData> initAndGetMetricDataObserver() { start(); return new SerializedObserver<>(metricDataSubject); } WorkerMetricHandler(final String jobId,
                               final Observer<JobAutoScaler.Event> jobAutoScaleObserver,
                               final MantisMasterClientApi masterClientApi,
                               final AutoScaleMetricsConfig autoScaleMetricsConfig); Observer<MetricData> initAndGetMetricDataObserver();  }