@Test public void testMetricsPublisher() { final CompletableFuture<PutMetricDataResponse> putResponseFuture = new CompletableFuture<>(); putResponseFuture.complete(PutMetricDataResponse.builder().build()); when(cloudWatchClient.putMetricData(any(PutMetricDataRequest.class))).thenReturn(putResponseFuture); List<MetricDatumWithKey<CloudWatchMetricKey>> dataToPublish = constructMetricDatumWithKeyList(25); List<Map<String, MetricDatum>> expectedData = constructMetricDatumListMap(dataToPublish); publisher.publishMetrics(dataToPublish); ArgumentCaptor<PutMetricDataRequest> argument = ArgumentCaptor.forClass(PutMetricDataRequest.class); Mockito.verify(cloudWatchClient, Mockito.atLeastOnce()).putMetricData(argument.capture()); List<PutMetricDataRequest> requests = argument.getAllValues(); Assert.assertEquals(expectedData.size(), requests.size()); for (int i = 0; i < requests.size(); i++) { assertMetricData(expectedData.get(i), requests.get(i)); } }
public void publishMetrics(List<MetricDatumWithKey<CloudWatchMetricKey>> dataToPublish) { for (int startIndex = 0; startIndex < dataToPublish.size(); startIndex += BATCH_SIZE) { int endIndex = Math.min(dataToPublish.size(), startIndex + BATCH_SIZE); PutMetricDataRequest.Builder request = PutMetricDataRequest.builder(); request = request.namespace(namespace); List<MetricDatum> metricData = new ArrayList<>(); for (int i = startIndex; i < endIndex; i++) { metricData.add(dataToPublish.get(i).datum); } request = request.metricData(metricData); try { PutMetricDataRequest.Builder finalRequest = request; blockingExecute(cloudWatchAsyncClient.putMetricData(finalRequest.build()), PUT_TIMEOUT_MILLIS, CW_EXCEPTION_MANAGER); } catch(CloudWatchException | TimeoutException e) { log.warn("Could not publish {} datums to CloudWatch", endIndex - startIndex, e); } catch (Exception e) { log.error("Unknown exception while publishing {} datums to CloudWatch", endIndex - startIndex, e); } } }
CloudWatchMetricsPublisher { public void publishMetrics(List<MetricDatumWithKey<CloudWatchMetricKey>> dataToPublish) { for (int startIndex = 0; startIndex < dataToPublish.size(); startIndex += BATCH_SIZE) { int endIndex = Math.min(dataToPublish.size(), startIndex + BATCH_SIZE); PutMetricDataRequest.Builder request = PutMetricDataRequest.builder(); request = request.namespace(namespace); List<MetricDatum> metricData = new ArrayList<>(); for (int i = startIndex; i < endIndex; i++) { metricData.add(dataToPublish.get(i).datum); } request = request.metricData(metricData); try { PutMetricDataRequest.Builder finalRequest = request; blockingExecute(cloudWatchAsyncClient.putMetricData(finalRequest.build()), PUT_TIMEOUT_MILLIS, CW_EXCEPTION_MANAGER); } catch(CloudWatchException | TimeoutException e) { log.warn("Could not publish {} datums to CloudWatch", endIndex - startIndex, e); } catch (Exception e) { log.error("Unknown exception while publishing {} datums to CloudWatch", endIndex - startIndex, e); } } } }
CloudWatchMetricsPublisher { public void publishMetrics(List<MetricDatumWithKey<CloudWatchMetricKey>> dataToPublish) { for (int startIndex = 0; startIndex < dataToPublish.size(); startIndex += BATCH_SIZE) { int endIndex = Math.min(dataToPublish.size(), startIndex + BATCH_SIZE); PutMetricDataRequest.Builder request = PutMetricDataRequest.builder(); request = request.namespace(namespace); List<MetricDatum> metricData = new ArrayList<>(); for (int i = startIndex; i < endIndex; i++) { metricData.add(dataToPublish.get(i).datum); } request = request.metricData(metricData); try { PutMetricDataRequest.Builder finalRequest = request; blockingExecute(cloudWatchAsyncClient.putMetricData(finalRequest.build()), PUT_TIMEOUT_MILLIS, CW_EXCEPTION_MANAGER); } catch(CloudWatchException | TimeoutException e) { log.warn("Could not publish {} datums to CloudWatch", endIndex - startIndex, e); } catch (Exception e) { log.error("Unknown exception while publishing {} datums to CloudWatch", endIndex - startIndex, e); } } } CloudWatchMetricsPublisher(CloudWatchAsyncClient cloudWatchClient, String namespace); }
CloudWatchMetricsPublisher { public void publishMetrics(List<MetricDatumWithKey<CloudWatchMetricKey>> dataToPublish) { for (int startIndex = 0; startIndex < dataToPublish.size(); startIndex += BATCH_SIZE) { int endIndex = Math.min(dataToPublish.size(), startIndex + BATCH_SIZE); PutMetricDataRequest.Builder request = PutMetricDataRequest.builder(); request = request.namespace(namespace); List<MetricDatum> metricData = new ArrayList<>(); for (int i = startIndex; i < endIndex; i++) { metricData.add(dataToPublish.get(i).datum); } request = request.metricData(metricData); try { PutMetricDataRequest.Builder finalRequest = request; blockingExecute(cloudWatchAsyncClient.putMetricData(finalRequest.build()), PUT_TIMEOUT_MILLIS, CW_EXCEPTION_MANAGER); } catch(CloudWatchException | TimeoutException e) { log.warn("Could not publish {} datums to CloudWatch", endIndex - startIndex, e); } catch (Exception e) { log.error("Unknown exception while publishing {} datums to CloudWatch", endIndex - startIndex, e); } } } CloudWatchMetricsPublisher(CloudWatchAsyncClient cloudWatchClient, String namespace); void publishMetrics(List<MetricDatumWithKey<CloudWatchMetricKey>> dataToPublish); }
CloudWatchMetricsPublisher { public void publishMetrics(List<MetricDatumWithKey<CloudWatchMetricKey>> dataToPublish) { for (int startIndex = 0; startIndex < dataToPublish.size(); startIndex += BATCH_SIZE) { int endIndex = Math.min(dataToPublish.size(), startIndex + BATCH_SIZE); PutMetricDataRequest.Builder request = PutMetricDataRequest.builder(); request = request.namespace(namespace); List<MetricDatum> metricData = new ArrayList<>(); for (int i = startIndex; i < endIndex; i++) { metricData.add(dataToPublish.get(i).datum); } request = request.metricData(metricData); try { PutMetricDataRequest.Builder finalRequest = request; blockingExecute(cloudWatchAsyncClient.putMetricData(finalRequest.build()), PUT_TIMEOUT_MILLIS, CW_EXCEPTION_MANAGER); } catch(CloudWatchException | TimeoutException e) { log.warn("Could not publish {} datums to CloudWatch", endIndex - startIndex, e); } catch (Exception e) { log.error("Unknown exception while publishing {} datums to CloudWatch", endIndex - startIndex, e); } } } CloudWatchMetricsPublisher(CloudWatchAsyncClient cloudWatchClient, String namespace); void publishMetrics(List<MetricDatumWithKey<CloudWatchMetricKey>> dataToPublish);  }