@Test(expected = InternalErrorException.class) public void failedProcessWhenSerializerThrowError() { Request request = new Request(HttpPut.METHOD_NAME, TestConstants.DUMMY_ENDPOINT); request.setJsonEntity(TestConstants.CUSTOMER_INGESTION_PAYLOAD_NESTED); IngestionSerializer mockBadSerializer = mock(IngestionSerializer.class); test = new IndexProcessor(mockBadSerializer, mockComprehendClient, mockESClient, mockConfigRetriever, mockExecutorService); when(HTTPTransformer.transformHttpEntityToString(any(HttpEntity.class))) .thenReturn(TestConstants.CUSTOMER_INGESTION_PAYLOAD_NESTED); when(mockBadSerializer.deserialize(anyString())) .thenThrow(new InternalErrorException(TestConstants.DUMMY_ERROR_MESSAGE)); test.processRequest(request, mockLogger); verifyZeroInteractions(mockExecutorService); }
@Override public Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger) { logger.log("Index request detected"); String payloadStr = HTTPTransformer.transformHttpEntityToString(request.getEntity()); Map<String, ComprehendConfiguration> configMap = configRetriever.retrieveStoredConfig(); String indexName = RequestIdentifier.getIndexName(request); if (Strings.isNullOrEmpty(payloadStr) || configMap.isEmpty()) { logger.log("The Payload or Comprehend config is empty, return pass through requests"); return esClient.performRequest(request); } try { JsonNode inputJson = ingestionSerializer.deserialize(payloadStr); List<Callable<SingularResponse>> callableList = createCallableList(indexName, inputJson, configMap); if (callableList.isEmpty()) { logger.log("No config field detected, return pass through requests"); return esClient.performRequest(request); } List<Future<SingularResponse>> executionResult = executorService.invokeAll(callableList, Constants.INDEX_EXECUTOR_TIMEOUT_SECONDS, TimeUnit.SECONDS); JSONObject payloadJson = new JSONObject(payloadStr); for (Future<SingularResponse> result : executionResult) { try { SingularResponse responseObject = result.get(); payloadJson.put(responseObject.getFieldNameAndOperation(), removeResponseMetadata(responseObject.getComprehendResult())); JSONObject flattenedResponse = responseObject.getFlattenedResult(); if (flattenedResponse != null) { payloadJson.put(String.format("%s_%s", responseObject.getFieldNameAndOperation(), Constants.KIBANA_KEY_NAME), removeResponseMetadata(flattenedResponse)); } } catch (ExecutionException | InterruptedException e) { logger.log("Got exception when retrieving the comprehend response: " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR); } } payloadJson.put(Constants.TIME_STAMP_KEY, Instant.now().toString()); logger.log("Ingest Comprehend enriched results to Elasticsearch"); return esClient.performRequest(request.getMethod(), request.getEndpoint(), payloadJson.toString()); } catch (Exception e) { logger.log("Exceptions happen when trying to process index request. " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR, e); } }
IndexProcessor implements ElasticsearchProcessor { @Override public Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger) { logger.log("Index request detected"); String payloadStr = HTTPTransformer.transformHttpEntityToString(request.getEntity()); Map<String, ComprehendConfiguration> configMap = configRetriever.retrieveStoredConfig(); String indexName = RequestIdentifier.getIndexName(request); if (Strings.isNullOrEmpty(payloadStr) || configMap.isEmpty()) { logger.log("The Payload or Comprehend config is empty, return pass through requests"); return esClient.performRequest(request); } try { JsonNode inputJson = ingestionSerializer.deserialize(payloadStr); List<Callable<SingularResponse>> callableList = createCallableList(indexName, inputJson, configMap); if (callableList.isEmpty()) { logger.log("No config field detected, return pass through requests"); return esClient.performRequest(request); } List<Future<SingularResponse>> executionResult = executorService.invokeAll(callableList, Constants.INDEX_EXECUTOR_TIMEOUT_SECONDS, TimeUnit.SECONDS); JSONObject payloadJson = new JSONObject(payloadStr); for (Future<SingularResponse> result : executionResult) { try { SingularResponse responseObject = result.get(); payloadJson.put(responseObject.getFieldNameAndOperation(), removeResponseMetadata(responseObject.getComprehendResult())); JSONObject flattenedResponse = responseObject.getFlattenedResult(); if (flattenedResponse != null) { payloadJson.put(String.format("%s_%s", responseObject.getFieldNameAndOperation(), Constants.KIBANA_KEY_NAME), removeResponseMetadata(flattenedResponse)); } } catch (ExecutionException | InterruptedException e) { logger.log("Got exception when retrieving the comprehend response: " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR); } } payloadJson.put(Constants.TIME_STAMP_KEY, Instant.now().toString()); logger.log("Ingest Comprehend enriched results to Elasticsearch"); return esClient.performRequest(request.getMethod(), request.getEndpoint(), payloadJson.toString()); } catch (Exception e) { logger.log("Exceptions happen when trying to process index request. " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR, e); } } }
IndexProcessor implements ElasticsearchProcessor { @Override public Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger) { logger.log("Index request detected"); String payloadStr = HTTPTransformer.transformHttpEntityToString(request.getEntity()); Map<String, ComprehendConfiguration> configMap = configRetriever.retrieveStoredConfig(); String indexName = RequestIdentifier.getIndexName(request); if (Strings.isNullOrEmpty(payloadStr) || configMap.isEmpty()) { logger.log("The Payload or Comprehend config is empty, return pass through requests"); return esClient.performRequest(request); } try { JsonNode inputJson = ingestionSerializer.deserialize(payloadStr); List<Callable<SingularResponse>> callableList = createCallableList(indexName, inputJson, configMap); if (callableList.isEmpty()) { logger.log("No config field detected, return pass through requests"); return esClient.performRequest(request); } List<Future<SingularResponse>> executionResult = executorService.invokeAll(callableList, Constants.INDEX_EXECUTOR_TIMEOUT_SECONDS, TimeUnit.SECONDS); JSONObject payloadJson = new JSONObject(payloadStr); for (Future<SingularResponse> result : executionResult) { try { SingularResponse responseObject = result.get(); payloadJson.put(responseObject.getFieldNameAndOperation(), removeResponseMetadata(responseObject.getComprehendResult())); JSONObject flattenedResponse = responseObject.getFlattenedResult(); if (flattenedResponse != null) { payloadJson.put(String.format("%s_%s", responseObject.getFieldNameAndOperation(), Constants.KIBANA_KEY_NAME), removeResponseMetadata(flattenedResponse)); } } catch (ExecutionException | InterruptedException e) { logger.log("Got exception when retrieving the comprehend response: " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR); } } payloadJson.put(Constants.TIME_STAMP_KEY, Instant.now().toString()); logger.log("Ingest Comprehend enriched results to Elasticsearch"); return esClient.performRequest(request.getMethod(), request.getEndpoint(), payloadJson.toString()); } catch (Exception e) { logger.log("Exceptions happen when trying to process index request. " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR, e); } }  }
IndexProcessor implements ElasticsearchProcessor { @Override public Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger) { logger.log("Index request detected"); String payloadStr = HTTPTransformer.transformHttpEntityToString(request.getEntity()); Map<String, ComprehendConfiguration> configMap = configRetriever.retrieveStoredConfig(); String indexName = RequestIdentifier.getIndexName(request); if (Strings.isNullOrEmpty(payloadStr) || configMap.isEmpty()) { logger.log("The Payload or Comprehend config is empty, return pass through requests"); return esClient.performRequest(request); } try { JsonNode inputJson = ingestionSerializer.deserialize(payloadStr); List<Callable<SingularResponse>> callableList = createCallableList(indexName, inputJson, configMap); if (callableList.isEmpty()) { logger.log("No config field detected, return pass through requests"); return esClient.performRequest(request); } List<Future<SingularResponse>> executionResult = executorService.invokeAll(callableList, Constants.INDEX_EXECUTOR_TIMEOUT_SECONDS, TimeUnit.SECONDS); JSONObject payloadJson = new JSONObject(payloadStr); for (Future<SingularResponse> result : executionResult) { try { SingularResponse responseObject = result.get(); payloadJson.put(responseObject.getFieldNameAndOperation(), removeResponseMetadata(responseObject.getComprehendResult())); JSONObject flattenedResponse = responseObject.getFlattenedResult(); if (flattenedResponse != null) { payloadJson.put(String.format("%s_%s", responseObject.getFieldNameAndOperation(), Constants.KIBANA_KEY_NAME), removeResponseMetadata(flattenedResponse)); } } catch (ExecutionException | InterruptedException e) { logger.log("Got exception when retrieving the comprehend response: " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR); } } payloadJson.put(Constants.TIME_STAMP_KEY, Instant.now().toString()); logger.log("Ingest Comprehend enriched results to Elasticsearch"); return esClient.performRequest(request.getMethod(), request.getEndpoint(), payloadJson.toString()); } catch (Exception e) { logger.log("Exceptions happen when trying to process index request. " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR, e); } }  @Override Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger); }
IndexProcessor implements ElasticsearchProcessor { @Override public Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger) { logger.log("Index request detected"); String payloadStr = HTTPTransformer.transformHttpEntityToString(request.getEntity()); Map<String, ComprehendConfiguration> configMap = configRetriever.retrieveStoredConfig(); String indexName = RequestIdentifier.getIndexName(request); if (Strings.isNullOrEmpty(payloadStr) || configMap.isEmpty()) { logger.log("The Payload or Comprehend config is empty, return pass through requests"); return esClient.performRequest(request); } try { JsonNode inputJson = ingestionSerializer.deserialize(payloadStr); List<Callable<SingularResponse>> callableList = createCallableList(indexName, inputJson, configMap); if (callableList.isEmpty()) { logger.log("No config field detected, return pass through requests"); return esClient.performRequest(request); } List<Future<SingularResponse>> executionResult = executorService.invokeAll(callableList, Constants.INDEX_EXECUTOR_TIMEOUT_SECONDS, TimeUnit.SECONDS); JSONObject payloadJson = new JSONObject(payloadStr); for (Future<SingularResponse> result : executionResult) { try { SingularResponse responseObject = result.get(); payloadJson.put(responseObject.getFieldNameAndOperation(), removeResponseMetadata(responseObject.getComprehendResult())); JSONObject flattenedResponse = responseObject.getFlattenedResult(); if (flattenedResponse != null) { payloadJson.put(String.format("%s_%s", responseObject.getFieldNameAndOperation(), Constants.KIBANA_KEY_NAME), removeResponseMetadata(flattenedResponse)); } } catch (ExecutionException | InterruptedException e) { logger.log("Got exception when retrieving the comprehend response: " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR); } } payloadJson.put(Constants.TIME_STAMP_KEY, Instant.now().toString()); logger.log("Ingest Comprehend enriched results to Elasticsearch"); return esClient.performRequest(request.getMethod(), request.getEndpoint(), payloadJson.toString()); } catch (Exception e) { logger.log("Exceptions happen when trying to process index request. " + e); throw new InternalErrorException(CustomerMessage.INTERNAL_ERROR, e); } }  @Override Response processRequest(@NonNull final Request request, @NonNull final LambdaLogger logger); final static String SDK_RESPONSE_METADATA_KEY; final static String SDK_HTTP_METADATA_KEY; }