@Test public void testExecuteIncreasingLatencyCheckEnabled() throws InterruptedException { this.countDownLatch = new CountDownLatch(1); Config config = ConfigFactory.empty().withValue(KafkaIngestionHealthCheck.KAFKA_INGESTION_HEALTH_CHECK_EXPECTED_CONSUMPTION_RATE_MBPS_KEY, ConfigValueFactory.fromAnyRef(5)) .withValue(KafkaIngestionHealthCheck.KAFKA_INGESTION_HEALTH_CHECK_LATENCY_THRESHOLD_MINUTES_KEY, ConfigValueFactory.fromAnyRef(5)); KafkaExtractorStatsTracker extractorStatsTracker = Mockito.mock(KafkaExtractorStatsTracker.class); Mockito.when(extractorStatsTracker.getMaxIngestionLatency(TimeUnit.MINUTES)) .thenReturn(6L) .thenReturn(7L) .thenReturn(7L) .thenReturn(5L); Mockito.when(extractorStatsTracker.getConsumptionRateMBps()) .thenReturn(2.0) .thenReturn(1.5) .thenReturn(2.1) .thenReturn(2.5); KafkaIngestionHealthCheck check = new KafkaIngestionHealthCheck(config, extractorStatsTracker); check.execute(); this.countDownLatch.await(10, TimeUnit.MILLISECONDS); Assert.assertEquals(this.countDownLatch.getCount(), 1L); check.execute(); this.countDownLatch.await(10, TimeUnit.MILLISECONDS); Assert.assertEquals(this.countDownLatch.getCount(), 1L); check.execute(); this.countDownLatch.await(10, TimeUnit.MILLISECONDS); Assert.assertEquals(this.countDownLatch.getCount(), 0); this.countDownLatch = new CountDownLatch(1); check.execute(); this.countDownLatch.await(10, TimeUnit.MILLISECONDS); Assert.assertEquals(this.countDownLatch.getCount(), 1); config = config.withValue(KafkaIngestionHealthCheck.KAFKA_INGESTION_HEALTH_CHECK_INCREASING_LATENCY_CHECK_ENABLED_KEY, ConfigValueFactory.fromAnyRef(false)); extractorStatsTracker = Mockito.mock(KafkaExtractorStatsTracker.class); Mockito.when(extractorStatsTracker.getMaxIngestionLatency(TimeUnit.MINUTES)) .thenReturn(10L) .thenReturn(7L) .thenReturn(5L); Mockito.when(extractorStatsTracker.getConsumptionRateMBps()) .thenReturn(2.0) .thenReturn(1.5) .thenReturn(2.1); check = new KafkaIngestionHealthCheck(config, extractorStatsTracker); check.execute(); }
@Override public void execute() { this.ingestionLatencies.add(this.statsTracker.getMaxIngestionLatency(TimeUnit.MINUTES)); this.consumptionRateMBps.add(this.statsTracker.getConsumptionRateMBps()); if (ingestionLatencies.size() < this.slidingWindowSize) { log.info("SUCCESS: Num observations: {} smaller than {}", ingestionLatencies.size(), this.slidingWindowSize); return; } if (!checkIngestionLatency()) { log.info("SUCCESS: Ingestion Latencies = {}, Ingestion Latency Threshold: {}", this.ingestionLatencies.toString(), this.ingestionLatencyThresholdMinutes); return; } double avgConsumptionRate = getMaxConsumptionRate(); if (avgConsumptionRate > this.consumptionRateDropOffFraction * this.expectedConsumptionRate) { log.info("SUCCESS: Avg. Consumption Rate = {} MBps, Target Consumption rate = {} MBps", avgConsumptionRate, this.expectedConsumptionRate); return; } log.error("FAILED: {}", getHealthCheckReport()); if (this.eventBus != null) { log.info("Posting {} message to EventBus", ContainerHealthCheckFailureEvent.class.getSimpleName()); ContainerHealthCheckFailureEvent event = new ContainerHealthCheckFailureEvent(this.config, getClass().getName()); event.addMetadata("ingestionLatencies", this.ingestionLatencies.toString()); event.addMetadata("consumptionRates", this.consumptionRateMBps.toString()); event.addMetadata("ingestionLatencyThreshold", Long.toString(this.ingestionLatencyThresholdMinutes)); event.addMetadata("targetConsumptionRate", Double.toString(this.expectedConsumptionRate)); this.eventBus.post(event); } }
KafkaIngestionHealthCheck implements CommitStep { @Override public void execute() { this.ingestionLatencies.add(this.statsTracker.getMaxIngestionLatency(TimeUnit.MINUTES)); this.consumptionRateMBps.add(this.statsTracker.getConsumptionRateMBps()); if (ingestionLatencies.size() < this.slidingWindowSize) { log.info("SUCCESS: Num observations: {} smaller than {}", ingestionLatencies.size(), this.slidingWindowSize); return; } if (!checkIngestionLatency()) { log.info("SUCCESS: Ingestion Latencies = {}, Ingestion Latency Threshold: {}", this.ingestionLatencies.toString(), this.ingestionLatencyThresholdMinutes); return; } double avgConsumptionRate = getMaxConsumptionRate(); if (avgConsumptionRate > this.consumptionRateDropOffFraction * this.expectedConsumptionRate) { log.info("SUCCESS: Avg. Consumption Rate = {} MBps, Target Consumption rate = {} MBps", avgConsumptionRate, this.expectedConsumptionRate); return; } log.error("FAILED: {}", getHealthCheckReport()); if (this.eventBus != null) { log.info("Posting {} message to EventBus", ContainerHealthCheckFailureEvent.class.getSimpleName()); ContainerHealthCheckFailureEvent event = new ContainerHealthCheckFailureEvent(this.config, getClass().getName()); event.addMetadata("ingestionLatencies", this.ingestionLatencies.toString()); event.addMetadata("consumptionRates", this.consumptionRateMBps.toString()); event.addMetadata("ingestionLatencyThreshold", Long.toString(this.ingestionLatencyThresholdMinutes)); event.addMetadata("targetConsumptionRate", Double.toString(this.expectedConsumptionRate)); this.eventBus.post(event); } } }
KafkaIngestionHealthCheck implements CommitStep { @Override public void execute() { this.ingestionLatencies.add(this.statsTracker.getMaxIngestionLatency(TimeUnit.MINUTES)); this.consumptionRateMBps.add(this.statsTracker.getConsumptionRateMBps()); if (ingestionLatencies.size() < this.slidingWindowSize) { log.info("SUCCESS: Num observations: {} smaller than {}", ingestionLatencies.size(), this.slidingWindowSize); return; } if (!checkIngestionLatency()) { log.info("SUCCESS: Ingestion Latencies = {}, Ingestion Latency Threshold: {}", this.ingestionLatencies.toString(), this.ingestionLatencyThresholdMinutes); return; } double avgConsumptionRate = getMaxConsumptionRate(); if (avgConsumptionRate > this.consumptionRateDropOffFraction * this.expectedConsumptionRate) { log.info("SUCCESS: Avg. Consumption Rate = {} MBps, Target Consumption rate = {} MBps", avgConsumptionRate, this.expectedConsumptionRate); return; } log.error("FAILED: {}", getHealthCheckReport()); if (this.eventBus != null) { log.info("Posting {} message to EventBus", ContainerHealthCheckFailureEvent.class.getSimpleName()); ContainerHealthCheckFailureEvent event = new ContainerHealthCheckFailureEvent(this.config, getClass().getName()); event.addMetadata("ingestionLatencies", this.ingestionLatencies.toString()); event.addMetadata("consumptionRates", this.consumptionRateMBps.toString()); event.addMetadata("ingestionLatencyThreshold", Long.toString(this.ingestionLatencyThresholdMinutes)); event.addMetadata("targetConsumptionRate", Double.toString(this.expectedConsumptionRate)); this.eventBus.post(event); } } KafkaIngestionHealthCheck(Config config, KafkaExtractorStatsTracker statsTracker); }
KafkaIngestionHealthCheck implements CommitStep { @Override public void execute() { this.ingestionLatencies.add(this.statsTracker.getMaxIngestionLatency(TimeUnit.MINUTES)); this.consumptionRateMBps.add(this.statsTracker.getConsumptionRateMBps()); if (ingestionLatencies.size() < this.slidingWindowSize) { log.info("SUCCESS: Num observations: {} smaller than {}", ingestionLatencies.size(), this.slidingWindowSize); return; } if (!checkIngestionLatency()) { log.info("SUCCESS: Ingestion Latencies = {}, Ingestion Latency Threshold: {}", this.ingestionLatencies.toString(), this.ingestionLatencyThresholdMinutes); return; } double avgConsumptionRate = getMaxConsumptionRate(); if (avgConsumptionRate > this.consumptionRateDropOffFraction * this.expectedConsumptionRate) { log.info("SUCCESS: Avg. Consumption Rate = {} MBps, Target Consumption rate = {} MBps", avgConsumptionRate, this.expectedConsumptionRate); return; } log.error("FAILED: {}", getHealthCheckReport()); if (this.eventBus != null) { log.info("Posting {} message to EventBus", ContainerHealthCheckFailureEvent.class.getSimpleName()); ContainerHealthCheckFailureEvent event = new ContainerHealthCheckFailureEvent(this.config, getClass().getName()); event.addMetadata("ingestionLatencies", this.ingestionLatencies.toString()); event.addMetadata("consumptionRates", this.consumptionRateMBps.toString()); event.addMetadata("ingestionLatencyThreshold", Long.toString(this.ingestionLatencyThresholdMinutes)); event.addMetadata("targetConsumptionRate", Double.toString(this.expectedConsumptionRate)); this.eventBus.post(event); } } KafkaIngestionHealthCheck(Config config, KafkaExtractorStatsTracker statsTracker); @Override boolean isCompleted(); @Override void execute(); }
KafkaIngestionHealthCheck implements CommitStep { @Override public void execute() { this.ingestionLatencies.add(this.statsTracker.getMaxIngestionLatency(TimeUnit.MINUTES)); this.consumptionRateMBps.add(this.statsTracker.getConsumptionRateMBps()); if (ingestionLatencies.size() < this.slidingWindowSize) { log.info("SUCCESS: Num observations: {} smaller than {}", ingestionLatencies.size(), this.slidingWindowSize); return; } if (!checkIngestionLatency()) { log.info("SUCCESS: Ingestion Latencies = {}, Ingestion Latency Threshold: {}", this.ingestionLatencies.toString(), this.ingestionLatencyThresholdMinutes); return; } double avgConsumptionRate = getMaxConsumptionRate(); if (avgConsumptionRate > this.consumptionRateDropOffFraction * this.expectedConsumptionRate) { log.info("SUCCESS: Avg. Consumption Rate = {} MBps, Target Consumption rate = {} MBps", avgConsumptionRate, this.expectedConsumptionRate); return; } log.error("FAILED: {}", getHealthCheckReport()); if (this.eventBus != null) { log.info("Posting {} message to EventBus", ContainerHealthCheckFailureEvent.class.getSimpleName()); ContainerHealthCheckFailureEvent event = new ContainerHealthCheckFailureEvent(this.config, getClass().getName()); event.addMetadata("ingestionLatencies", this.ingestionLatencies.toString()); event.addMetadata("consumptionRates", this.consumptionRateMBps.toString()); event.addMetadata("ingestionLatencyThreshold", Long.toString(this.ingestionLatencyThresholdMinutes)); event.addMetadata("targetConsumptionRate", Double.toString(this.expectedConsumptionRate)); this.eventBus.post(event); } } KafkaIngestionHealthCheck(Config config, KafkaExtractorStatsTracker statsTracker); @Override boolean isCompleted(); @Override void execute(); static final String KAFKA_INGESTION_HEALTH_CHECK_PREFIX; static final String KAFKA_INGESTION_HEALTH_CHECK_SLIDING_WINDOW_SIZE_KEY; static final String KAFKA_INGESTION_HEALTH_CHECK_LATENCY_THRESHOLD_MINUTES_KEY; static final String KAFKA_INGESTION_HEALTH_CHECK_CONSUMPTION_RATE_DROPOFF_FRACTION_KEY; static final String KAFKA_INGESTION_HEALTH_CHECK_EXPECTED_CONSUMPTION_RATE_MBPS_KEY; static final String KAFKA_INGESTION_HEALTH_CHECK_INCREASING_LATENCY_CHECK_ENABLED_KEY; static final int DEFAULT_KAFKA_INGESTION_HEALTH_CHECK_SLIDING_WINDOW_SIZE; static final long DEFAULT_KAFKA_INGESTION_HEALTH_CHECK_LATENCY_THRESHOLD_MINUTES; static final double DEFAULT_KAFKA_INGESTION_HEALTH_CHECK_CONSUMPTION_RATE_DROPOFF_FRACTION; static final double DEFAULT_KAFKA_INGESTION_HEALTH_CHECK_EXPECTED_CONSUMPTION_RATE_MBPS; }