thomasmueller commented on code in PR #2607:
URL: https://github.com/apache/jackrabbit-oak/pull/2607#discussion_r2495194252
##########
oak-run/src/main/java/org/apache/jackrabbit/oak/run/Downloader.java:
##########
@@ -170,29 +183,86 @@ public ItemResponse call() throws Exception {
Path destinationPath = Paths.get(item.destination);
Files.createDirectories(destinationPath.getParent());
+ long segmentSize = MAX_LENGTH_SINGLE_THREADED;
long size = 0;
- try (InputStream inputStream = sourceUrl.getInputStream();
- FileOutputStream outputStream = new
FileOutputStream(destinationPath.toFile())) {
- byte[] buffer = new byte[bufferSize];
- int bytesRead;
- while ((bytesRead = inputStream.read(buffer)) != -1) {
- if (md != null) {
- md.update(buffer, 0, bytesRead);
+ if (item.length >= segmentSize) {
+ size = item.length;
+ LOG.info("Downloading large file {}: {} bytes",
destinationPath.toString(), item.length);
+ String fileName = destinationPath.getFileName().toString();
+ long numSegments = (item.length + segmentSize - 1) /
segmentSize;
+ ArrayList<Path> segmentFiles = new ArrayList<>();
+ ArrayList<Future<Boolean>> downloadTasks = new ArrayList<>();
+ for (int i = 0; i < numSegments; i++) {
+ long startByte = i * segmentSize;
+ long endByte = Math.min(startByte + segmentSize - 1,
item.length - 1);
+ Path segmentFile =
destinationPath.getParent().resolve(fileName + "_" + i + ".tmp");
+ segmentFiles.add(segmentFile);
+ downloadTasks.add(executorService.submit(
+ new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ Exception lastException = null;
+ for (int i = 0; i < maxRetries; i++) {
+ try {
+ return
tryDownloadRange(item.source, connectTimeoutMs, readTimeoutMs,
+ segmentFile, startByte,
endByte);
+ } catch (Exception e) {
+ LOG.warn("Range download try # {}
failed", i, e);
+ lastException = e;
+ // retry
+ }
+ }
+ throw lastException;
+ }
+ }
+ ));
+ }
+ // wait for threads
+ boolean allSuccess = true;
+ for (int i = 0; i < downloadTasks.size(); i++) {
+ try {
+ boolean success = downloadTasks.get(i).get();
+ if (!success) {
+ allSuccess = false;
+ break;
+ }
+ } catch (Exception e) {
+ allSuccess = false;
+ break;
+ }
+ }
+ // merge
+ if (allSuccess) {
+ try (OutputStream fileOut =
Files.newOutputStream(destinationPath)) {
+ OutputStream out = md == null ? fileOut : new
DigestOutputStream(fileOut, md);
+ for (Path segmentFile : segmentFiles) {
+ if (Files.exists(segmentFile)) {
+ Files.copy(segmentFile, out);
+ Files.delete(segmentFile);
+ }
+ }
+ LOG.info("Download {} size {}, {} parts",
destinationPath.toString(), size, downloadTasks.size());
Review Comment:
You are right, it's no longer needed. I changed it to debug now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
