@Test public void testKafkaLagAndUserDefinedTriggersAutoScale() throws InterruptedException { final String jobId = "test-job-1"; final int stage = 1; final int workerIdx = 0; final int workerNum = 1; final int kafkaLag = 10_000; final int numWorkers = 2; final int metricValue = 1_000; final String testMetricGroup = "testMetricGroup"; final String testMetricName = "testMetricName"; final List<GaugeMeasurement> worker1KafkaGauges = Arrays.asList( new GaugeMeasurement(MetricStringConstants.KAFKA_LAG, kafkaLag / 2)); final List<GaugeMeasurement> worker2KafkaGauges = Arrays.asList( new GaugeMeasurement(MetricStringConstants.KAFKA_LAG, kafkaLag)); final List<GaugeMeasurement> gauges1 = Arrays.asList( new GaugeMeasurement(testMetricName, metricValue / 2)); final List<GaugeMeasurement> gauges2 = Arrays.asList( new GaugeMeasurement(testMetricName, metricValue)); final MantisMasterClientApi mockMasterClientApi = mock(MantisMasterClientApi.class); final Map<Integer, WorkerAssignments> assignmentsMap = new HashMap<>(); final Map<Integer, WorkerHost> hosts = new HashMap<>(); hosts.put(1, new WorkerHost("localhost", workerIdx, Arrays.asList(31300), MantisJobState.Started, workerNum, 31301, -1)); hosts.put(2, new WorkerHost("localhost", workerIdx + 1, Arrays.asList(31305), MantisJobState.Started, workerNum + 1, 31316, -1)); assignmentsMap.put(stage, new WorkerAssignments(stage, numWorkers, hosts)); when(mockMasterClientApi.schedulingChanges(jobId)).thenReturn(Observable.just(new JobSchedulingInfo(jobId, assignmentsMap))); final CountDownLatch latch = new CountDownLatch(2); final AutoScaleMetricsConfig aggregationConfig = new AutoScaleMetricsConfig(Collections.singletonMap(testMetricGroup, Collections.singletonMap(testMetricName, AutoScaleMetricsConfig.AggregationAlgo.AVERAGE))); 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); final long count = latch.getCount(); if (count == 2) { JobAutoScaler.Event expected1 = new JobAutoScaler.Event(StageScalingPolicy.ScalingReason.UserDefined, stage, metricValue * 3 / 4, numWorkers, ""); assertEquals(expected1, event); latch.countDown(); } if (count == 1) { JobAutoScaler.Event expected2 = new JobAutoScaler.Event(StageScalingPolicy.ScalingReason.KafkaLag, stage, kafkaLag, numWorkers, ""); assertEquals(expected2, event); latch.countDown(); } } }, mockMasterClientApi, aggregationConfig); final Observer<MetricData> metricDataObserver = workerMetricHandler.initAndGetMetricDataObserver(); metricDataObserver.onNext(new MetricData(jobId, stage, workerIdx, workerNum, KAFKA_CONSUMER_FETCH_MGR_METRIC_GROUP, worker1KafkaGauges)); metricDataObserver.onNext(new MetricData(jobId, stage, workerIdx + 1, workerNum + 1, KAFKA_CONSUMER_FETCH_MGR_METRIC_GROUP, worker2KafkaGauges)); metricDataObserver.onNext(new MetricData(jobId, stage, workerIdx, workerNum, testMetricGroup, gauges1)); metricDataObserver.onNext(new MetricData(jobId, stage, workerIdx + 1, workerNum + 1, testMetricGroup, gauges2)); 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();  }