@Test public void succeedPassThroughWhenPayloadDoNotContainIngestionRequest() { Request request = new Request(HttpPut.METHOD_NAME, TestConstants.DUMMY_ENDPOINT); request.setJsonEntity(TestConstants.CUSTOMER_PAYLOAD_NO_INGESTION_REQUEST); when(HTTPTransformer.transformHttpEntityToString(any(HttpEntity.class))) .thenReturn(TestConstants.CUSTOMER_PAYLOAD_NO_INGESTION_REQUEST); test.processRequest(request, mockLogger); verifyZeroInteractions(mockExecutorService); verify(mockLogger, times(1)) .log("No ingestion requests detected, return pass through requests"); }
@Override public Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger) { logger.log("Bulk requests detected"); String payloadStr = HTTPTransformer.transformHttpEntityToString(request.getEntity()); Map<String, ComprehendConfiguration> configMap = configRetriever.retrieveStoredConfig(); if (Strings.isNullOrEmpty(payloadStr) || configMap.isEmpty()) { logger.log("Payload or Comprehend config is empty, return pass through requests"); return esClient.performRequest(request); } List<String> payloadList = Arrays.asList(payloadStr.split(System.lineSeparator())); Map<Integer, BulkPayload> payloadMap = getIngestionPayloadMap(payloadList); if (payloadMap.isEmpty()) { logger.log("No ingestion requests detected, return pass through requests"); return esClient.performRequest(request); } List<BulkRequest> bulkRequestList = getBatchObjectList(payloadMap, configMap); if (bulkRequestList.isEmpty()) { logger.log("No config field was detected in the bulk request, return pass through requests"); return esClient.performRequest(request); } List<Callable<BatchResponse>> callableList = createCallableList(bulkRequestList); try { List<Future<BatchResponse>> executionResult = executorService.invokeAll(callableList, Constants.BULK_EXECUTOR_TIMEOUT_SECONDS, TimeUnit.SECONDS); for (Future<BatchResponse> result : executionResult) { try { attachComprehendResponse(payloadList, result, logger); logger.log("Extended original payload with Comprehend result"); } catch (RuntimeException e) { throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR); } } Request transformedRequest = new Request(request.getMethod(), request.getEndpoint()); transformedRequest.setJsonEntity(String.join(System.lineSeparator(), payloadList) + System.lineSeparator()); logger.log("Ingest Comprehend enriched bulk results to Elasticsearch"); Response response = esClient.performRequest(transformedRequest); logger.log(String.format("Response of bulk request: statusCode = %d, reasonPhrase = %s", response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase())); return response; } catch (InterruptedException | JSONException e) { logger.log("InternalErrorException happened when trying to process bulk request. " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR, e); } }
BulkProcessor implements ElasticsearchProcessor { @Override public Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger) { logger.log("Bulk requests detected"); String payloadStr = HTTPTransformer.transformHttpEntityToString(request.getEntity()); Map<String, ComprehendConfiguration> configMap = configRetriever.retrieveStoredConfig(); if (Strings.isNullOrEmpty(payloadStr) || configMap.isEmpty()) { logger.log("Payload or Comprehend config is empty, return pass through requests"); return esClient.performRequest(request); } List<String> payloadList = Arrays.asList(payloadStr.split(System.lineSeparator())); Map<Integer, BulkPayload> payloadMap = getIngestionPayloadMap(payloadList); if (payloadMap.isEmpty()) { logger.log("No ingestion requests detected, return pass through requests"); return esClient.performRequest(request); } List<BulkRequest> bulkRequestList = getBatchObjectList(payloadMap, configMap); if (bulkRequestList.isEmpty()) { logger.log("No config field was detected in the bulk request, return pass through requests"); return esClient.performRequest(request); } List<Callable<BatchResponse>> callableList = createCallableList(bulkRequestList); try { List<Future<BatchResponse>> executionResult = executorService.invokeAll(callableList, Constants.BULK_EXECUTOR_TIMEOUT_SECONDS, TimeUnit.SECONDS); for (Future<BatchResponse> result : executionResult) { try { attachComprehendResponse(payloadList, result, logger); logger.log("Extended original payload with Comprehend result"); } catch (RuntimeException e) { throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR); } } Request transformedRequest = new Request(request.getMethod(), request.getEndpoint()); transformedRequest.setJsonEntity(String.join(System.lineSeparator(), payloadList) + System.lineSeparator()); logger.log("Ingest Comprehend enriched bulk results to Elasticsearch"); Response response = esClient.performRequest(transformedRequest); logger.log(String.format("Response of bulk request: statusCode = %d, reasonPhrase = %s", response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase())); return response; } catch (InterruptedException | JSONException e) { logger.log("InternalErrorException happened when trying to process bulk request. " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR, e); } } }
BulkProcessor implements ElasticsearchProcessor { @Override public Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger) { logger.log("Bulk requests detected"); String payloadStr = HTTPTransformer.transformHttpEntityToString(request.getEntity()); Map<String, ComprehendConfiguration> configMap = configRetriever.retrieveStoredConfig(); if (Strings.isNullOrEmpty(payloadStr) || configMap.isEmpty()) { logger.log("Payload or Comprehend config is empty, return pass through requests"); return esClient.performRequest(request); } List<String> payloadList = Arrays.asList(payloadStr.split(System.lineSeparator())); Map<Integer, BulkPayload> payloadMap = getIngestionPayloadMap(payloadList); if (payloadMap.isEmpty()) { logger.log("No ingestion requests detected, return pass through requests"); return esClient.performRequest(request); } List<BulkRequest> bulkRequestList = getBatchObjectList(payloadMap, configMap); if (bulkRequestList.isEmpty()) { logger.log("No config field was detected in the bulk request, return pass through requests"); return esClient.performRequest(request); } List<Callable<BatchResponse>> callableList = createCallableList(bulkRequestList); try { List<Future<BatchResponse>> executionResult = executorService.invokeAll(callableList, Constants.BULK_EXECUTOR_TIMEOUT_SECONDS, TimeUnit.SECONDS); for (Future<BatchResponse> result : executionResult) { try { attachComprehendResponse(payloadList, result, logger); logger.log("Extended original payload with Comprehend result"); } catch (RuntimeException e) { throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR); } } Request transformedRequest = new Request(request.getMethod(), request.getEndpoint()); transformedRequest.setJsonEntity(String.join(System.lineSeparator(), payloadList) + System.lineSeparator()); logger.log("Ingest Comprehend enriched bulk results to Elasticsearch"); Response response = esClient.performRequest(transformedRequest); logger.log(String.format("Response of bulk request: statusCode = %d, reasonPhrase = %s", response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase())); return response; } catch (InterruptedException | JSONException e) { logger.log("InternalErrorException happened when trying to process bulk request. " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR, e); } }  }
BulkProcessor implements ElasticsearchProcessor { @Override public Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger) { logger.log("Bulk requests detected"); String payloadStr = HTTPTransformer.transformHttpEntityToString(request.getEntity()); Map<String, ComprehendConfiguration> configMap = configRetriever.retrieveStoredConfig(); if (Strings.isNullOrEmpty(payloadStr) || configMap.isEmpty()) { logger.log("Payload or Comprehend config is empty, return pass through requests"); return esClient.performRequest(request); } List<String> payloadList = Arrays.asList(payloadStr.split(System.lineSeparator())); Map<Integer, BulkPayload> payloadMap = getIngestionPayloadMap(payloadList); if (payloadMap.isEmpty()) { logger.log("No ingestion requests detected, return pass through requests"); return esClient.performRequest(request); } List<BulkRequest> bulkRequestList = getBatchObjectList(payloadMap, configMap); if (bulkRequestList.isEmpty()) { logger.log("No config field was detected in the bulk request, return pass through requests"); return esClient.performRequest(request); } List<Callable<BatchResponse>> callableList = createCallableList(bulkRequestList); try { List<Future<BatchResponse>> executionResult = executorService.invokeAll(callableList, Constants.BULK_EXECUTOR_TIMEOUT_SECONDS, TimeUnit.SECONDS); for (Future<BatchResponse> result : executionResult) { try { attachComprehendResponse(payloadList, result, logger); logger.log("Extended original payload with Comprehend result"); } catch (RuntimeException e) { throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR); } } Request transformedRequest = new Request(request.getMethod(), request.getEndpoint()); transformedRequest.setJsonEntity(String.join(System.lineSeparator(), payloadList) + System.lineSeparator()); logger.log("Ingest Comprehend enriched bulk results to Elasticsearch"); Response response = esClient.performRequest(transformedRequest); logger.log(String.format("Response of bulk request: statusCode = %d, reasonPhrase = %s", response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase())); return response; } catch (InterruptedException | JSONException e) { logger.log("InternalErrorException happened when trying to process bulk request. " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR, e); } }  @Override Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger); }
BulkProcessor implements ElasticsearchProcessor { @Override public Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger) { logger.log("Bulk requests detected"); String payloadStr = HTTPTransformer.transformHttpEntityToString(request.getEntity()); Map<String, ComprehendConfiguration> configMap = configRetriever.retrieveStoredConfig(); if (Strings.isNullOrEmpty(payloadStr) || configMap.isEmpty()) { logger.log("Payload or Comprehend config is empty, return pass through requests"); return esClient.performRequest(request); } List<String> payloadList = Arrays.asList(payloadStr.split(System.lineSeparator())); Map<Integer, BulkPayload> payloadMap = getIngestionPayloadMap(payloadList); if (payloadMap.isEmpty()) { logger.log("No ingestion requests detected, return pass through requests"); return esClient.performRequest(request); } List<BulkRequest> bulkRequestList = getBatchObjectList(payloadMap, configMap); if (bulkRequestList.isEmpty()) { logger.log("No config field was detected in the bulk request, return pass through requests"); return esClient.performRequest(request); } List<Callable<BatchResponse>> callableList = createCallableList(bulkRequestList); try { List<Future<BatchResponse>> executionResult = executorService.invokeAll(callableList, Constants.BULK_EXECUTOR_TIMEOUT_SECONDS, TimeUnit.SECONDS); for (Future<BatchResponse> result : executionResult) { try { attachComprehendResponse(payloadList, result, logger); logger.log("Extended original payload with Comprehend result"); } catch (RuntimeException e) { throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR); } } Request transformedRequest = new Request(request.getMethod(), request.getEndpoint()); transformedRequest.setJsonEntity(String.join(System.lineSeparator(), payloadList) + System.lineSeparator()); logger.log("Ingest Comprehend enriched bulk results to Elasticsearch"); Response response = esClient.performRequest(transformedRequest); logger.log(String.format("Response of bulk request: statusCode = %d, reasonPhrase = %s", response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase())); return response; } catch (InterruptedException | JSONException e) { logger.log("InternalErrorException happened when trying to process bulk request. " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR, e); } }  @Override Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger);  }