Skip to content
Snippets Groups Projects
Commit b27d2a78 authored by MIchael Nguyen's avatar MIchael Nguyen
Browse files

updating queue

parent c0d7e6fd
No related branches found
No related tags found
1 merge request!6Trusted ibm
Hide whitespace changes
Inline Side-by-side
provider/indexer-aws/indexer-queue-aws/src/main/java/org/opengroup/osdu/indexerqueue/aws/api/IndexerQueue.java
+ 23
− 23
View file @ b27d2a78
...@@ -43,13 +43,13 @@ public class IndexerQueue { ...@@ -43,13 +43,13 @@ public class IndexerQueue {
public static void main(String[] args) { public static void main(String[] args) {
EnvironmentVariables environmentVariables = new EnvironmentVariables(); EnvironmentVariables environmentVariables = new EnvironmentVariables();
AmazonLogConfig logConfig = new AmazonLogConfig(environmentVariables.region); // AmazonLogConfig logConfig = new AmazonLogConfig(environmentVariables.region);
AWSLogs log = logConfig.AmazonLog(); // AWSLogs log = logConfig.AmazonLog();
PutLogEventsRequest logRequest = new PutLogEventsRequest("ECSLogGroup-indexer-queue", "indexer-queue", new ArrayList<>()); // PutLogEventsRequest logRequest = new PutLogEventsRequest("ECSLogGroup-indexer-queue", "indexer-queue", new ArrayList<>());
InputLogEvent logEvent = new InputLogEvent().withMessage("Starting Indexer Queue and obtaining Arguments"); // InputLogEvent logEvent = new InputLogEvent().withMessage("Starting Indexer Queue and obtaining Arguments");
logRequest.getLogEvents().add(logEvent); // logRequest.getLogEvents().add(logEvent);
logRequest.getLogEvents().add(new InputLogEvent().withMessage("Retrieving indexer service account JWT")); // logRequest.getLogEvents().add(new InputLogEvent().withMessage("Retrieving indexer service account JWT"));
System.out.println(environmentVariables.region); System.out.println(environmentVariables.region);
System.out.println(environmentVariables.maxBatchRequestCount); System.out.println(environmentVariables.maxBatchRequestCount);
System.out.println(environmentVariables.queueName); System.out.println(environmentVariables.queueName);
...@@ -64,66 +64,66 @@ public class IndexerQueue { ...@@ -64,66 +64,66 @@ public class IndexerQueue {
System.out.println(environmentVariables.keepAliveTimeInMin); System.out.println(environmentVariables.keepAliveTimeInMin);
try { try {
logRequest.getLogEvents().add(new InputLogEvent().withMessage("Starting Indexer Queue and obtaining Arguments")); // logRequest.getLogEvents().add(new InputLogEvent().withMessage("Starting Indexer Queue and obtaining Arguments"));
logRequest.getLogEvents().add(new InputLogEvent().withMessage("Retrieving indexer service account JWT")); // logRequest.getLogEvents().add(new InputLogEvent().withMessage("Retrieving indexer service account JWT"));
AWSCognitoClient cognitoClient = new AWSCognitoClient(environmentVariables.cognitoClientId, environmentVariables.cognitoAuthFlow, AWSCognitoClient cognitoClient = new AWSCognitoClient(environmentVariables.cognitoClientId, environmentVariables.cognitoAuthFlow,
environmentVariables.cognitoUser, environmentVariables.cognitoPassword); environmentVariables.cognitoUser, environmentVariables.cognitoPassword);
String indexerServiceAccountJWT = cognitoClient.getToken(); String indexerServiceAccountJWT = cognitoClient.getToken();
if(indexerServiceAccountJWT == null){ if(indexerServiceAccountJWT == null){
logRequest.getLogEvents().add(new InputLogEvent().withMessage("Indexer service account not set up correctly")); // logRequest.getLogEvents().add(new InputLogEvent().withMessage("Indexer service account not set up correctly"));
} }
logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("Connecting to the SQS Queue: %s", environmentVariables.queueName))); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("Connecting to the SQS Queue: %s", environmentVariables.queueName)));
AmazonSQSConfig sqsConfig = new AmazonSQSConfig(environmentVariables.region); AmazonSQSConfig sqsConfig = new AmazonSQSConfig(environmentVariables.region);
AmazonSQS sqsClient = sqsConfig.AmazonSQS(); AmazonSQS sqsClient = sqsConfig.AmazonSQS();
logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("Creating a thread pool with %s threads", environmentVariables.maxIndexThreads))); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("Creating a thread pool with %s threads", environmentVariables.maxIndexThreads)));
ThreadPoolExecutor executorPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(environmentVariables.maxIndexThreads); ThreadPoolExecutor executorPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(environmentVariables.maxIndexThreads);
final String deadLetterQueueUrl = sqsClient.getQueueUrl(environmentVariables.deadLetterQueueName).getQueueUrl(); final String deadLetterQueueUrl = sqsClient.getQueueUrl(environmentVariables.deadLetterQueueName).getQueueUrl();
System.out.println(deadLetterQueueUrl); System.out.println(deadLetterQueueUrl);
List<Message> messages = IndexerQueueService.getMessages(sqsClient, environmentVariables.queueName, environmentVariables.maxBatchRequestCount, environmentVariables.maxMessagesAllowed); List<Message> messages = IndexerQueueService.getMessages(sqsClient, environmentVariables.queueName, environmentVariables.maxBatchRequestCount, environmentVariables.maxMessagesAllowed);
logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("Processing %s messages from storage queue", messages.size()))); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("Processing %s messages from storage queue", messages.size())));
if (!messages.isEmpty()) { if (!messages.isEmpty()) {
List<IndexProcessor> indexProcessors = IndexerQueueService.processQueue(messages, environmentVariables.targetURL, executorPool, indexerServiceAccountJWT); List<IndexProcessor> indexProcessors = IndexerQueueService.processQueue(messages, environmentVariables.targetURL, executorPool, indexerServiceAccountJWT);
logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Messages Processed", indexProcessors.size()))); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Messages Processed", indexProcessors.size())));
List<IndexProcessor> failedProcessors = indexProcessors.stream().filter(indexProcessor -> indexProcessor.result == CallableResult.Fail || indexProcessor.exception != null).collect(Collectors.toList()); List<IndexProcessor> failedProcessors = indexProcessors.stream().filter(indexProcessor -> indexProcessor.result == CallableResult.Fail || indexProcessor.exception != null).collect(Collectors.toList());
logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Messages Failed", failedProcessors.size()))); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Messages Failed", failedProcessors.size())));
List<SendMessageResult> deadLetterResults = IndexerQueueService.sendMsgsToDeadLetterQueue(deadLetterQueueUrl, failedProcessors, sqsClient); List<SendMessageResult> deadLetterResults = IndexerQueueService.sendMsgsToDeadLetterQueue(deadLetterQueueUrl, failedProcessors, sqsClient);
logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Messages Dead Lettered", deadLetterResults.size()))); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Messages Dead Lettered", deadLetterResults.size())));
List<DeleteMessageBatchRequestEntry> deleteEntries = indexProcessors.stream().map(indexProcessor -> new DeleteMessageBatchRequestEntry(indexProcessor.messageId, indexProcessor.receiptHandle)).collect(Collectors.toList()); List<DeleteMessageBatchRequestEntry> deleteEntries = indexProcessors.stream().map(indexProcessor -> new DeleteMessageBatchRequestEntry(indexProcessor.messageId, indexProcessor.receiptHandle)).collect(Collectors.toList());
logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Messages Deleting", deleteEntries.size()))); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Messages Deleting", deleteEntries.size())));
final String sqsQueueUrl = sqsClient.getQueueUrl(environmentVariables.queueName).getQueueUrl(); final String sqsQueueUrl = sqsClient.getQueueUrl(environmentVariables.queueName).getQueueUrl();
List<DeleteMessageBatchRequest> deleteBatchRequests = IndexerQueueService.createMultipleBatchDeleteRequest(sqsQueueUrl, deleteEntries, environmentVariables.maxBatchRequestCount); List<DeleteMessageBatchRequest> deleteBatchRequests = IndexerQueueService.createMultipleBatchDeleteRequest(sqsQueueUrl, deleteEntries, environmentVariables.maxBatchRequestCount);
logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Delete Batch Request Created", deleteBatchRequests.size()))); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Delete Batch Request Created", deleteBatchRequests.size())));
List<DeleteMessageBatchResult> deleteMessageBatchResults = IndexerQueueService.deleteMessages(deleteBatchRequests, sqsClient); List<DeleteMessageBatchResult> deleteMessageBatchResults = IndexerQueueService.deleteMessages(deleteBatchRequests, sqsClient);
logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Requests Deleted", deleteMessageBatchResults.size()))); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(String.format("%s Requests Deleted", deleteMessageBatchResults.size())));
} }
} catch (ExecutionException e) { } catch (ExecutionException e) {
System.out.println(e); System.out.println(e);
logRequest.getLogEvents().add(new InputLogEvent().withMessage(e.getMessage())); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(e.getMessage()));
} catch (InterruptedException e) { } catch (InterruptedException e) {
System.out.println(e); System.out.println(e);
logRequest.getLogEvents().add(new InputLogEvent().withMessage(e.getMessage())); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(e.getMessage()));
} catch (NullPointerException e) { } catch (NullPointerException e) {
System.out.println(e); System.out.println(e);
logRequest.getLogEvents().add(new InputLogEvent().withMessage(e.getMessage())); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(e.getMessage()));
}catch (Exception e) { }catch (Exception e) {
System.out.println(e); System.out.println(e);
logRequest.getLogEvents().add(new InputLogEvent().withMessage(e.getMessage())); // logRequest.getLogEvents().add(new InputLogEvent().withMessage(e.getMessage()));
} finally { } finally {
System.out.println("failing finally"); System.out.println("failing finally");
log.putLogEvents(logRequest); // log.putLogEvents(logRequest);
} }
} }
} }
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment