From 3a28eefaf6bcbe60c3bb0d1f885119217577fba6 Mon Sep 17 00:00:00 2001 From: Rory Torney Date: Tue, 30 Jun 2026 18:14:45 +0100 Subject: [PATCH 1/3] wip --- .../staging/dao/filesystem/FileSystemDao.java | 10 +++-- .../services/staging/utils/JsonMinifier.java | 37 ++++++++++++++++--- 2 files changed, 39 insertions(+), 8 deletions(-) diff --git a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java index 4da9f8b9..be4af439 100644 --- a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java +++ b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java @@ -229,9 +229,13 @@ private void cleanupInProgressBatch(final File inProgressBatchFolderPath) private void completeInProgressBatch(final TenantId tenantId, final Path inProgressBatchFolderPath, final BatchId batchId) throws StagingException { - LOGGER.info("Completing batch with id {} for {}...", batchId, tenantId); - final Path batchFolder = batchPathProvider.getPathForBatch(tenantId, batchId); + LOGGER.info( + "Completing batch with id {} for {}. Moving in-progress path {} to completed path {}.", + batchId, + tenantId, + inProgressBatchFolderPath, + batchFolder); if (batchFolder.toFile().exists()) { LOGGER.warn("Batch {} has been previously uploaded. Removing previously uploaded batch...", batchId); try { @@ -251,7 +255,7 @@ private void completeInProgressBatch(final TenantId tenantId, final Path inProgr throw new StagingException(ex); } - LOGGER.debug("Batch {} completed successfully.", batchId); + LOGGER.info("Batch {} completed successfully for {} at {}.", batchId, tenantId, batchFolder); } /** diff --git a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java index d661ab51..94e76f7f 100644 --- a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java +++ b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java @@ -28,6 +28,7 @@ import java.io.InputStream; import java.io.OutputStream; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.util.Map; @@ -110,6 +111,7 @@ private static void processJsonTokens(final JsonParser parser, boolean updateReference = false; // mark if the updateReference is for a file that has been uploaded boolean isLocalRefFile = false; + String currentDocumentFieldName = null; while ((token = parser.nextToken()) != null) { switch (token) { case FIELD_NAME: @@ -126,6 +128,7 @@ private static void processJsonTokens(final JsonParser parser, bufferData = false; bufferEncoding = false; pauseWriting = false; + currentDocumentFieldName = parser.getText(); } break; case VALUE_FALSE: @@ -157,7 +160,13 @@ private static void processJsonTokens(final JsonParser parser, if (dataBuffer != null && dataBuffer.getBytes().length > fieldValueSizeThreshold) { // write it out to a loose file final String fileName = RandomStringUtils.randomAlphanumeric(10); - final String contentFileName = writeDataToFile(dataBuffer, fileName, inprogressContentFolderPath, encodingBuffer); + final String contentFileName = writeDataToFile( + dataBuffer, + fileName, + inprogressContentFolderPath, + encodingBuffer, + currentDocumentFieldName, + fieldValueSizeThreshold); dataBuffer = contentFileName; encodingBuffer = STORAGE_REF; updateReference = true; @@ -205,21 +214,39 @@ private static void processJsonTokens(final JsonParser parser, } } - private static String writeDataToFile(final String data, final String fileName, - final String inprogressContentFolderPath, final String encoding) throws IOException + private static String writeDataToFile(final String data, + final String fileName, + final String inprogressContentFolderPath, + final String encoding, + final String documentFieldName, + final int fieldValueSizeThreshold) throws IOException { String contentFileName = fileName; + Path targetFile = null; + long inputBytes = data.getBytes(StandardCharsets.UTF_8).length; if (encoding == null || encoding.equalsIgnoreCase(UTF8_ENCODING)) { contentFileName = fileName + TXT_ENTENSION; - final Path targetFile = Paths.get(inprogressContentFolderPath, contentFileName); + targetFile = Paths.get(inprogressContentFolderPath, contentFileName); FileUtils.writeStringToFile(targetFile.toFile(), data, StandardCharsets.UTF_8); } else if (encoding.equalsIgnoreCase(BASE64_ENCODING)) { contentFileName = fileName + BINARY_ENTENSION; - final Path targetFile = Paths.get(inprogressContentFolderPath, contentFileName); + targetFile = Paths.get(inprogressContentFolderPath, contentFileName); // If encoding is base64, write file after base64 decoding the data final byte[] decodedData = Base64.decodeBase64(data); + inputBytes = decodedData.length; FileUtils.writeByteArrayToFile(targetFile.toFile(), decodedData); } + if (targetFile != null) { + LOGGER.info( + "Staging service externalized JSON field data to storage_ref. Field: {}; Encoding: {}; " + + "Input bytes: {}; File bytes: {}; Threshold bytes: {}; File: {}", + documentFieldName, + encoding == null ? UTF8_ENCODING : encoding, + inputBytes, + Files.size(targetFile), + fieldValueSizeThreshold, + targetFile); + } return contentFileName; } } From 696c446adfa7de43f242769ba88893596e2dc2e8 Mon Sep 17 00:00:00 2001 From: Rory Torney Date: Tue, 30 Jun 2026 21:02:30 +0100 Subject: [PATCH 2/3] wip --- .../staging/dao/filesystem/FileSystemDao.java | 49 ++++++++++++++++++- .../dao/filesystem/SubBatchWriter.java | 13 +++++ .../services/staging/utils/JsonMinifier.java | 34 +++++++++++++ 3 files changed, 94 insertions(+), 2 deletions(-) diff --git a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java index be4af439..b1acfe7a 100644 --- a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java +++ b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java @@ -134,7 +134,14 @@ public void deleteBatch(final TenantId tenantId, final BatchId batchId) } try { + LOGGER.info( + "Deleting completed batch with id {} for {}. Batch bytes: {}; Path: {}", + batchId, + tenantId, + calculateDirectorySize(batchPath), + batchPath); FileUtils.deleteDirectory(batchPath.toFile()); + LOGGER.info("Deleted completed batch with id {} for {} at {}.", batchId, tenantId, batchPath); } catch (IOException ex) { throw new StagingException(ex); } @@ -192,7 +199,14 @@ public List saveFiles(final TenantId tenantId, final BatchId batchId, fi LOGGER.debug("Reading loose file..."); try (final InputStream inStream = fileItemInput.getInputStream()) { FileUtils.copyInputStreamToFile(inStream, targetFile); - LOGGER.debug("Loose file written to {}", targetFile); + LOGGER.info( + "Staging service stored loose file part. Tenant: {}; Batch: {}; Field name: {}; " + + "Stored file: {}; File bytes: {}", + tenantId, + batchId, + filename, + targetFile, + Files.size(targetFile.toPath())); fileNames.add(targetFileName); binaryFilesUploaded.put(filename, targetFileName); } catch (IOException ex) { @@ -255,7 +269,38 @@ private void completeInProgressBatch(final TenantId tenantId, final Path inProgr throw new StagingException(ex); } - LOGGER.info("Batch {} completed successfully for {} at {}.", batchId, tenantId, batchFolder); + LOGGER.info( + "Batch {} completed successfully for {} at {}. Batch bytes: {}", + batchId, + tenantId, + batchFolder, + calculateDirectorySize(batchFolder)); + } + + private static long calculateDirectorySize(final Path directory) + { + try (final Stream paths = Files.walk(directory, FileVisitOption.FOLLOW_LINKS)) { + return paths + .filter(Files::isRegularFile) + .mapToLong(FileSystemDao::getFileSizeSafely) + .sum(); + } catch (final IOException ex) { + LOGGER.info( + "Unable to calculate staged batch size. Directory: {}; Error: {}", + directory, + ex.getMessage()); + return 0; + } + } + + private static long getFileSizeSafely(final Path file) + { + try { + return Files.size(file); + } catch (final IOException ex) { + LOGGER.info("Unable to read staged file size while calculating batch size. File: {}; Error: {}", file, ex.getMessage()); + return 0; + } } /** diff --git a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/SubBatchWriter.java b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/SubBatchWriter.java index 370c0d13..c079645f 100644 --- a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/SubBatchWriter.java +++ b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/SubBatchWriter.java @@ -46,6 +46,7 @@ public class SubBatchWriter implements AutoCloseable private final File inProgressBatchFolder; private final int subbatchSize; private OutputStream outStream; + private File currentSubBatch; //Track number of document files processed private int count = 0; @@ -60,6 +61,7 @@ private void createSubBatchOutStream() throws StagingException //Make a new subbatch file final String subBatchFileName = BatchNameProvider.getSubBatchName(); final File subBatch = inProgressBatchFolder.toPath().resolve(subBatchFileName).toFile(); + currentSubBatch = subBatch; LOGGER.debug("Created new subbatchFile : {} ", subBatch); //Open new stream to start writing to subbatch file @@ -91,10 +93,20 @@ public void writeDocumentFile(final InputStreamSupplier inputStreamSupplier, createSubBatchOutStream(); } + final long subBatchBytesBeforeWrite = currentSubBatch.length(); try (final InputStream inStream = inputStreamSupplier.get()) { try { JsonMinifier.validateAndMinifyJson(inStream, outStream, storageRefFolderPath, inprogressContentFolderPath, fieldValueSizeThreshold, binaryFilesUploaded); + outStream.flush(); + final long subBatchBytesAfterWrite = currentSubBatch.length(); + LOGGER.info( + "Staging service wrote document JSON to subbatch. Document number in subbatch: {}; " + + "Bytes written: {}; Subbatch bytes: {}; Subbatch file: {}", + count + 1, + subBatchBytesAfterWrite - subBatchBytesBeforeWrite, + subBatchBytesAfterWrite, + currentSubBatch); count++; } catch (final JsonProcessingException | InvalidDocumentException ex) { LOGGER.error("Error staging document", ex); @@ -120,6 +132,7 @@ public void close() throws Exception outStream.close(); } outStream = null; + currentSubBatch = null; } } diff --git a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java index 94e76f7f..f23c6e83 100644 --- a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java +++ b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java @@ -184,6 +184,8 @@ private static void processJsonTokens(final JsonParser parser, } else { dataBuffer = storageRefPath + "/" + dataBuffer; } + } else if (STORAGE_REF.equalsIgnoreCase(encodingBuffer)) { + logExistingStorageReference(dataBuffer, currentDocumentFieldName); } gen.writeFieldName(DATA_FIELD); gen.writeString(dataBuffer); @@ -249,4 +251,36 @@ private static String writeDataToFile(final String data, } return contentFileName; } + + private static void logExistingStorageReference(final String storageReference, final String documentFieldName) + { + if (storageReference == null) { + return; + } + + final Path storageReferencePath = Paths.get(storageReference); + if (Files.exists(storageReferencePath)) { + try { + LOGGER.info( + "Staging service observed existing storage_ref JSON field data. " + + "Field: {}; File bytes: {}; File: {}", + documentFieldName, + Files.size(storageReferencePath), + storageReferencePath); + } catch (final IOException ex) { + LOGGER.info( + "Staging service observed existing storage_ref JSON field data but could not read file size. " + + "Field: {}; File: {}; Error: {}", + documentFieldName, + storageReferencePath, + ex.getMessage()); + } + } else { + LOGGER.info( + "Staging service observed existing storage_ref JSON field data but the file was not present. " + + "Field: {}; File: {}", + documentFieldName, + storageReferencePath); + } + } } From 2d75b3f07bd9d275aa44ffedaacc1bbb1ec43a0b Mon Sep 17 00:00:00 2001 From: Rory Torney Date: Tue, 30 Jun 2026 21:04:06 +0100 Subject: [PATCH 3/3] wip --- .../staging/dao/filesystem/FileSystemDao.java | 5 ++++- .../services/staging/utils/JsonMinifier.java | 15 ++++++++++++++- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java index b1acfe7a..c440fad3 100644 --- a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java +++ b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/dao/filesystem/FileSystemDao.java @@ -298,7 +298,10 @@ private static long getFileSizeSafely(final Path file) try { return Files.size(file); } catch (final IOException ex) { - LOGGER.info("Unable to read staged file size while calculating batch size. File: {}; Error: {}", file, ex.getMessage()); + LOGGER.info( + "Unable to read staged file size while calculating batch size. File: {}; Error: {}", + file, + ex.getMessage()); return 0; } } diff --git a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java index f23c6e83..17a47902 100644 --- a/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java +++ b/staging-service/src/main/java/com/github/cafdataprocessing/services/staging/utils/JsonMinifier.java @@ -29,6 +29,7 @@ import java.io.OutputStream; import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.nio.file.InvalidPathException; import java.nio.file.Path; import java.nio.file.Paths; import java.util.Map; @@ -258,7 +259,19 @@ private static void logExistingStorageReference(final String storageReference, f return; } - final Path storageReferencePath = Paths.get(storageReference); + final Path storageReferencePath; + try { + storageReferencePath = Paths.get(storageReference); + } catch (final InvalidPathException ex) { + LOGGER.info( + "Staging service observed existing storage_ref JSON field data but the reference is not a local path. " + + "Field: {}; Reference: {}; Error: {}", + documentFieldName, + storageReference, + ex.getMessage()); + return; + } + if (Files.exists(storageReferencePath)) { try { LOGGER.info(