micord/ervu-eks
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix S3 calls

Browse source
This commit is contained in:
Maksim Tereshin 2024-11-24 14:43:54 +01:00
parent 771157abb1
commit adf30c5bf8
No known key found for this signature in database
1 changed files with 19 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

41
config-data-executor/src/main/java/org/micord/service/RequestService.java
View file

@ -42,24 +42,21 @@ public class RequestService {
public void processS3Requests(List<S3Request> s3Requests, List<String> ids) {
logger.debug("Starting processing of S3 requests");
if (s3Requests != null) {
s3Requests.forEach(request -> {
List<CompletableFuture<Void>> futures = ids.stream()
.map(id -> CompletableFuture.runAsync(() -> processS3Request(request, id)))
.toList();
if (s3Requests != null && !s3Requests.isEmpty()) {
List<CompletableFuture<Void>> requestFutures = s3Requests.stream()
.map(request -> CompletableFuture.runAsync(() -> processS3Request(request, ids)))
.toList();
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
.thenRun(() -> logger.info("Successfully processed all S3 requests."))
.exceptionally(ex -> {
logger.error("Failed to process S3 requests", ex);
return null;
});
});
CompletableFuture.allOf(requestFutures.toArray(new CompletableFuture[0]))
.thenRun(() -> logger.info("Successfully processed all S3 requests."))
.exceptionally(ex -> {
logger.error("Failed to process S3 requests", ex);
return null;
});
}
}
private void processS3Request(S3Request request, String id) {
logger.debug("Starting processing of S3 request for id: {}", id);
private void processS3Request(S3Request request, List<String> ids) {
try {
List<String> files = new ArrayList<>();
@ -67,10 +64,11 @@ public class RequestService {
for (RequestArgument argument : request.getRequestArguments()) {
try (Connection connection = DatabaseConnection.getConnection(
argument.getRequestArgumentConnectionParams())) {
String query = argument.getRequestArgumentURL();
Map<String, Object> query = buildSqlQuery(request, ids);
logger.debug("Starting fetching paths from database for S3 request");
long startExecTime = System.currentTimeMillis();
List<String> result = fetchFileListFromDatabaseSQL(connection, query);
List<String> result = fetchFileListFromDatabaseSQL(connection, (String) query.get("requestURL"));
if (result != null && !result.isEmpty()) {
files.addAll(result);
}
@ -100,23 +98,23 @@ public class RequestService {
.thenAccept(response -> {
if (response.statusCode() == HttpURLConnection.HTTP_NO_CONTENT
|| response.statusCode() == HttpURLConnection.HTTP_OK) {
logger.info("Successfully deleted object for ID {}", id);
logger.info("Successfully deleted object {}", file);
}
else {
logger.error("Failed to delete object for ID {}. Response code: {}", id,
logger.error("Failed to delete object {}. Response code: {}", file,
response.statusCode()
);
}
})
.exceptionally(ex -> {
logger.error("Failed to delete object for ID {}", id, ex);
logger.error("Failed to delete object {}", file, ex);
return null;
});
});
}
catch (Exception e) {
logger.error("Failed to process S3 request for id: {}", id, e);
logger.error("Failed to process S3 request: {}", request, e);
}
}
@ -163,7 +161,7 @@ public class RequestService {
}
}
private Map<String, Object> buildSqlQuery(SqlRequest request, List<String> ids) {
private Map<String, Object> buildSqlQuery(BaseRequest request, List<String> ids) {
logger.debug("Starting building SQL query for request: {}", request.getRequestURL());
long startExecTime = System.currentTimeMillis();
Map<String, Object> resultMap = new HashMap<>();
@ -212,7 +210,6 @@ public class RequestService {
}
resultMap.put("requestURL", requestURL
.replace("${DB}", request.getSqlConnectionParams().getJdbcDatabase())
.replace("${endpointArguments}", endpointArguments));
long endExecTime = System.currentTimeMillis();