Merge pull request #73 from rpmoore/master

Adding logging around various conditions in the helper functions

Author: rpmoore

ds3-sdk/src/main/java/com/spectralogic/ds3client/helpers/ChunkTransferrer.java

@@ -24,6 +24,8 @@
 import com.spectralogic.ds3client.models.bulk.Node;
 import com.spectralogic.ds3client.models.bulk.Objects;
 import com.spectralogic.ds3client.serializer.XmlProcessingException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.security.SignatureException;
@@ -33,6 +35,7 @@
 import java.util.concurrent.Executors;
 
 class ChunkTransferrer {
+    private final static Logger LOG = LoggerFactory.getLogger(ChunkTransferrer.class);
     private final ItemTransferrer itemTransferrer;
     private final Ds3Client mainClient;
     private final JobPartTracker partTracker;
@@ -57,18 +60,24 @@ public void transferChunks(
             final Iterable<Node> nodes,
             final Iterable<Objects> chunks)
                 throws SignatureException, IOException, XmlProcessingException {
+        LOG.debug("Getting ready to process chunks");
         final Map<UUID, Node> nodeMap = buildNodeMap(nodes);
+        LOG.debug("Starting executor service");
         final ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(maxParallelRequests));
+        LOG.debug("Executor service started");
         try {
             final List<ListenableFuture<?>> tasks = new ArrayList<>();
             for (final Objects chunk : chunks) {
+                LOG.debug("Processing parts for chunk: " + chunk.getChunkId().toString());
                 final Ds3Client client = mainClient.newForNode(nodeMap.get(chunk.getNodeId()));
                 for (final BulkObject ds3Object : chunk) {
                     final ObjectPart part = new ObjectPart(ds3Object.getOffset(), ds3Object.getLength());
                     if (this.partTracker.containsPart(ds3Object.getName(), part)) {
+                        LOG.debug("Adding " + ds3Object.getName() + " to executor for processing");
                         tasks.add(executor.submit(new Callable<Object>() {
                             @Override
                             public Object call() throws Exception {
+                                LOG.debug("Processing " + ds3Object.getName());
                                 ChunkTransferrer.this.itemTransferrer.transferItem(client, ds3Object);
                                 ChunkTransferrer.this.partTracker.completePart(ds3Object.getName(), part);
                                 return null;
@@ -79,6 +88,7 @@ public Object call() throws Exception {
             }
             executeWithExceptionHandling(tasks);
         } finally {
+            LOG.debug("Shutting down executor");
             executor.shutdown();
         }
     }

ds3-sdk/src/main/java/com/spectralogic/ds3client/helpers/WriteJobImpl.java

@@ -31,7 +31,6 @@
 import java.io.IOException;
 import java.security.SignatureException;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
@@ -41,11 +40,13 @@ public WriteJobImpl(
             final Ds3Client client,
             final MasterObjectList masterObjectList) {
         super(client, masterObjectList);
+        LOG.info("Ready to start transfer for job " + masterObjectList.getJobId().toString() + " with " + masterObjectList.getObjects().size() + " chunks");
     }
 
     @Override
     public void transfer(final ObjectChannelBuilder channelBuilder)
             throws SignatureException, IOException, XmlProcessingException {
+        LOG.debug("Starting job transfer");
         final List<Objects> filteredChunks = filterChunks(this.masterObjectList.getObjects());
         try (final JobState jobState = new JobState(channelBuilder, filteredChunks)) {
             final ChunkTransferrer chunkTransferrer = new ChunkTransferrer(
@@ -55,6 +56,7 @@ public void transfer(final ObjectChannelBuilder channelBuilder)
                 this.maxParallelRequests
             );
             for (final Objects chunk : filteredChunks) {
+                LOG.debug("Allocating chunk: " + chunk.getChunkId().toString());
                 chunkTransferrer.transferChunks(this.masterObjectList.getNodes(), Collections.singletonList(filterChunk(allocateChunk(chunk))));
             }
         } catch (final SignatureException | IOException | XmlProcessingException | RuntimeException e) {
@@ -75,12 +77,16 @@ private Objects allocateChunk(final Objects filtered) throws IOException, Signat
     private Objects tryAllocateChunk(final Objects filtered) throws IOException, SignatureException {
         final AllocateJobChunkResponse response =
                 this.client.allocateJobChunk(new AllocateJobChunkRequest(filtered.getChunkId()));
+
+        LOG.info("AllocatedJobChunkResponse status: " + response.getStatus().toString());
         switch (response.getStatus()) {
         case ALLOCATED:
             return response.getObjects();
         case RETRYLATER:
             try {
-                Thread.sleep(response.getRetryAfterSeconds() * 1000);
+                final int retryAfter = response.getRetryAfterSeconds() * 1000;
+                LOG.debug("Will retry allocate chunk call after " + retryAfter + " seconds");
+                Thread.sleep(retryAfter);
                 return null;
             } catch (final InterruptedException e) {
                 throw new RuntimeException(e);