Index: jackrabbit-aws-ext/src/main/java/org/apache/jackrabbit/aws/ext/ds/S3Backend.java
===================================================================
--- jackrabbit-aws-ext/src/main/java/org/apache/jackrabbit/aws/ext/ds/S3Backend.java	(revision 1579937)
+++ jackrabbit-aws-ext/src/main/java/org/apache/jackrabbit/aws/ext/ds/S3Backend.java	(working copy)
@@ -36,6 +36,7 @@
 import org.apache.jackrabbit.aws.ext.S3Constants;
 import org.apache.jackrabbit.aws.ext.Utils;
 import org.apache.jackrabbit.core.data.AsyncUploadCallback;
+import org.apache.jackrabbit.core.data.AsyncUploadResult;
 import org.apache.jackrabbit.core.data.Backend;
 import org.apache.jackrabbit.core.data.CachingDataStore;
 import org.apache.jackrabbit.core.data.DataIdentifier;
@@ -44,6 +45,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.amazonaws.AmazonClientException;
 import com.amazonaws.AmazonServiceException;
 import com.amazonaws.event.ProgressEvent;
 import com.amazonaws.event.ProgressListener;
@@ -99,6 +101,8 @@
     private Properties properties;
 
     private Date startTime;
+    
+    private ThreadPoolExecutor asyncWriteExecuter;
 
     /**
      * Initialize S3Backend. It creates AmazonS3Client and TransferManager from
@@ -135,9 +139,7 @@
             startTime = new Date();
             Thread.currentThread().setContextClassLoader(
                 getClass().getClassLoader());
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("init");
-            }
+            LOG.debug("init");
             this.store = store;
             s3service = Utils.openService(prop);
             if (bucket == null || "".equals(bucket.trim())) {
@@ -157,9 +159,9 @@
                     s3service.createBucket(bucket, region);
                     endpoint = S3 + DASH + region + DOT + AWSDOTCOM;
                 }
-                LOG.info("Created bucket: " + bucket + " in " + region);
+                LOG.info("Created bucket [{}] in [{}] ", bucket, region);
             } else {
-                LOG.info("Using bucket: " + bucket);
+                LOG.info("Using bucket [{}]", bucket);
                 if (DEFAULT_AWS_BUCKET_REGION.equals(region)) {
                     endpoint = S3 + DOT + AWSDOTCOM;
                 } else if (Region.EU_Ireland.toString().equals(region)) {
@@ -178,18 +180,27 @@
              * redirects it to correct location.
              */
             s3service.setEndpoint(endpoint);
-            LOG.info("S3 service endpoint: " + endpoint);
+            LOG.info("S3 service endpoint [{}] ", endpoint);
 
             int writeThreads = 10;
             String writeThreadsStr = prop.getProperty(S3Constants.S3_WRITE_THREADS);
             if (writeThreadsStr != null) {
                 writeThreads = Integer.parseInt(writeThreadsStr);
             }
-            LOG.info("Using thread pool of [" + writeThreads
-                + "] threads in S3 transfer manager");
+            LOG.info("Using thread pool of [{}] threads in S3 transfer manager.", writeThreads);
             tmx = new TransferManager(s3service,
                 (ThreadPoolExecutor) Executors.newFixedThreadPool(writeThreads,
                     new NamedThreadFactory("s3-transfer-manager-worker")));
+            
+            int asyncWritePoolSize = 10;
+            String maxConnsStr = prop.getProperty(S3Constants.S3_MAX_CONNS);
+            if (maxConnsStr != null) {
+                asyncWritePoolSize = Integer.parseInt(maxConnsStr)
+                    - writeThreads;
+            }
+            
+            asyncWriteExecuter = (ThreadPoolExecutor) Executors.newFixedThreadPool(
+                asyncWritePoolSize, new NamedThreadFactory("s3-write-worker"));
             String renameKeyProp = prop.getProperty(S3Constants.S3_RENAME_KEYS);
             boolean renameKeyBool = (renameKeyProp == null || "".equals(renameKeyProp))
                     ? true
@@ -197,15 +208,10 @@
             if (renameKeyBool) {
                 renameKeys();
             }
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("S3 Backend initialized in ["
-                    + (System.currentTimeMillis() - startTime.getTime())
-                    + "] ms");
-            }
+            LOG.debug("S3 Backend initialized in [{}] ms",
+                +(System.currentTimeMillis() - startTime.getTime()));
         } catch (Exception e) {
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("  error ", e);
-            }
+            LOG.debug("  error ", e);
             throw new DataStoreException("Could not initialize S3 from "
                 + prop, e);
         } finally {
@@ -233,8 +239,8 @@
             throw new IllegalArgumentException(
                 "callback parameter cannot be null in asyncUpload");
         }
-        Thread th = new Thread(new AsyncUploadJob(identifier, file, callback));
-        th.start();
+        asyncWriteExecuter.execute(new AsyncUploadJob(identifier, file,
+            callback));
     }
 
     /**
@@ -251,17 +257,15 @@
             ObjectMetadata objectMetaData = s3service.getObjectMetadata(bucket,
                 key);
             if (objectMetaData != null) {
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug("exists [" + identifier + "]: [true] took ["
-                        + (System.currentTimeMillis() - start) + "] ms");
-                }
+                LOG.debug("exists [{}]: [true] took [{}] ms.",
+                    identifier, (System.currentTimeMillis() - start) );
                 return true;
             }
             return false;
         } catch (AmazonServiceException e) {
             if (e.getStatusCode() == 404) {
-                LOG.info("exists [" + identifier + "]: [false] took ["
-                    + (System.currentTimeMillis() - start) + "] ms");
+                LOG.debug("exists [{}]: [false] took [{}] ms.",
+                    identifier, (System.currentTimeMillis() - start) );
                 return false;
             }
             throw new DataStoreException(
@@ -293,11 +297,8 @@
                         key, bucket, key);
                     copReq.setNewObjectMetadata(objectMetaData);
                     s3service.copyObject(copReq);
-                    if (LOG.isDebugEnabled()) {
-                        LOG.debug("[ " + identifier.toString()
-                            + "] touched took ["
-                            + (System.currentTimeMillis() - start) + "] ms");
-                    }
+                    LOG.debug("[{}] touched took [{}] ms. ", identifier,
+                        (System.currentTimeMillis() - start));
                 }
             } else {
                 retVal = false;
@@ -320,11 +321,8 @@
                 Thread.currentThread().setContextClassLoader(contextClassLoader);
             }
         }
-
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("exists [" + identifier + "]: [" + retVal + "] took ["
-                + (System.currentTimeMillis() - start) + "] ms");
-        }
+        LOG.debug("exists [{}]: [{}] took [{}] ms.", new Object[] { identifier,
+            retVal, (System.currentTimeMillis() - start) });
         return retVal;
     }
 
@@ -339,10 +337,8 @@
                 getClass().getClassLoader());
             S3Object object = s3service.getObject(bucket, key);
             InputStream in = object.getObjectContent();
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("[ " + identifier.toString() + "] read took ["
-                    + (System.currentTimeMillis() - start) + "] ms");
-            }
+            LOG.debug("[{}] read took [{}]ms", identifier,
+                (System.currentTimeMillis() - start));
             return in;
         } catch (AmazonServiceException e) {
             throw new DataStoreException("Object not found: " + key, e);
@@ -373,11 +369,8 @@
                 if (!prevObjectListing.isTruncated()) break;
                 prevObjectListing = s3service.listNextBatchOfObjects(prevObjectListing);
             }
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("getAllIdentifiers returned size [ " + ids.size()
-                    + "] took [" + (System.currentTimeMillis() - start)
-                    + "] ms");
-            }
+            LOG.debug("getAllIdentifiers returned size [{}] took [{}] ms.",
+                ids.size(), (System.currentTimeMillis() - start));
             return ids.iterator();
         } catch (AmazonServiceException e) {
             throw new DataStoreException("Could not list objects", e);
@@ -399,17 +392,16 @@
                 getClass().getClassLoader());
             ObjectMetadata object = s3service.getObjectMetadata(bucket, key);
             long lastModified = object.getLastModified().getTime();
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Identifier [" + identifier.toString()
-                    + "] 's lastModified = [" + lastModified + "] took ["
-                    + (System.currentTimeMillis() - start) + "] ms");
-            }
+            LOG.debug(
+                "Identifier [{}]'s lastModified = [{}] took [{}]ms.",
+                new Object[] { identifier, lastModified,
+                    (System.currentTimeMillis() - start) });
             return lastModified;
         } catch (AmazonServiceException e) {
             if (e.getStatusCode() == 404) {
-                LOG.info("getLastModified:Identifier [" + identifier.toString()
-                    + "] not found. Took ["
-                    + (System.currentTimeMillis() - start) + "]ms");
+                LOG.info(
+                    "getLastModified:Identifier [{}] not found. Took [{}] ms.",
+                    identifier, (System.currentTimeMillis() - start));
             }
             throw new DataStoreException(e);
         } finally {
@@ -429,11 +421,9 @@
                 getClass().getClassLoader());
             ObjectMetadata object = s3service.getObjectMetadata(bucket, key);
             long length = object.getContentLength();
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Identifier [" + identifier.toString()
-                    + "] 's length = [" + length + "] took ["
-                    + (System.currentTimeMillis() - start) + "] ms");
-            }
+            LOG.debug("Identifier [{}]'s length = [{}] took [{}]ms.",
+                new Object[] { identifier, length,
+                    (System.currentTimeMillis() - start) });
             return length;
         } catch (AmazonServiceException e) {
             throw new DataStoreException("Could not length of dataIdentifier "
@@ -455,11 +445,8 @@
             Thread.currentThread().setContextClassLoader(
                 getClass().getClassLoader());
             s3service.deleteObject(bucket, key);
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Identifier [" + identifier.toString()
-                    + "] 's deleted. It took ["
-                    + (System.currentTimeMillis() - start) + "] ms");
-            }
+            LOG.debug("Identifier [{}] deleted. It took [{}]ms.", new Object[] {
+                identifier, (System.currentTimeMillis() - start) });
         } catch (AmazonServiceException e) {
             throw new DataStoreException(
                 "Could not getLastModified of dataIdentifier " + identifier, e);
@@ -489,15 +476,9 @@
                     DataIdentifier identifier = new DataIdentifier(
                         getIdentifierName(s3ObjSumm.getKey()));
                     long lastModified = s3ObjSumm.getLastModified().getTime();
-                    if (LOG.isDebugEnabled()) {
-                        LOG.debug("id [" + identifier + "], lastModified ["
-                            + lastModified + "]");
-                    }
+                    LOG.debug("Identifier [{}]'s lastModified = [{}]", identifier, lastModified);
                     if (!store.isInUse(identifier) && lastModified < min) {
-                        if (LOG.isDebugEnabled()) {
-                            LOG.debug("add id :" + s3ObjSumm.getKey()
-                                + " to delete lists");
-                        }
+                        LOG.debug("add id [{}] to delete lists",  s3ObjSumm.getKey());
                         deleteList.add(new DeleteObjectsRequest.KeyVersion(
                             s3ObjSumm.getKey()));
                         deleteIdSet.add(identifier);
@@ -514,10 +495,8 @@
                                 + dobjs.getDeletedObjects().size() + " out of "
                                 + deleteList.size() + " are deleted");
                     } else {
-                        if (LOG.isDebugEnabled()) {
-                            LOG.debug(deleteList
-                                + " records deleted from datastore");
-                        }
+                        LOG.debug("[{}] records deleted from datastore",
+                            deleteList);
                     }
                 }
                 if (!prevObjectListing.isTruncated()) {
@@ -530,10 +509,10 @@
                 Thread.currentThread().setContextClassLoader(contextClassLoader);
             }
         }
-        LOG.info("deleteAllOlderThan: min=[" + min + "] exit. Deleted ["
-            + deleteIdSet + "] records. Number of records deleted ["
-            + deleteIdSet.size() + "] took ["
-            + (System.currentTimeMillis() - start) + "] ms");
+        LOG.info(
+            "deleteAllOlderThan: min=[{}] exit. Deleted[{}] records. Number of records deleted [{}] took [{}]ms",
+            new Object[] { min, deleteIdSet, deleteIdSet.size(),
+                (System.currentTimeMillis() - start) });
         return deleteIdSet;
     }
 
@@ -543,6 +522,7 @@
         tmx.abortMultipartUploads(bucket, startTime);
         tmx.shutdownNow();
         s3service.shutdown();
+        asyncWriteExecuter.shutdownNow();
         LOG.info("S3Backend closed.");
     }
 
@@ -588,21 +568,15 @@
                     throw new DataStoreException("Collision: " + key
                         + " new length: " + file.length() + " old length: " + l);
                 }
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug(key + "   exists, lastmodified ="
-                        + objectMetaData.getLastModified().getTime());
-                }
+                LOG.debug("[{}]'s exists, lastmodified = [{}]", key,
+                    objectMetaData.getLastModified().getTime());
                 CopyObjectRequest copReq = new CopyObjectRequest(bucket, key,
                     bucket, key);
                 copReq.setNewObjectMetadata(objectMetaData);
                 s3service.copyObject(copReq);
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug("lastModified of " + identifier.toString()
-                        + " updated successfully");
-                }
+                LOG.debug("lastModified of [{}] updated successfully.", identifier);
                 if (callback != null) {
-                    callback.call(identifier, file,
-                        AsyncUploadCallback.RESULT.SUCCESS);
+                    callback.onSuccess(new AsyncUploadResult(identifier, file));
                 }
             }
 
@@ -613,27 +587,22 @@
                         file));
                     // wait for upload to finish
                     if (asyncUpload) {
-                        up.addProgressListener(new S3UploadProgressListener(
+                        up.addProgressListener(new S3UploadProgressListener(up,
                             identifier, file, callback));
-                        if (LOG.isDebugEnabled()) {
-                            LOG.debug("added upload progress listener to identifier ["
-                                + identifier + "]");
-                        }
+                        LOG.debug(
+                            "added upload progress listener to identifier [{}]",
+                            identifier);
                     } else {
                         up.waitForUploadResult();
-                        if (LOG.isDebugEnabled()) {
-                            LOG.debug("synchronous upload to identifier ["
-                                + identifier + "] completed.");
-                        }
+                        LOG.debug("synchronous upload to identifier [{}] completed.", identifier); 
                         if (callback != null) {
-                            callback.call(identifier, file,
-                                AsyncUploadCallback.RESULT.SUCCESS);
+                            callback.onSuccess(new AsyncUploadResult(
+                                identifier, file));
                         }
                     }
                 } catch (Exception e2) {
                     if (!asyncUpload) {
-                        callback.call(identifier, file,
-                            AsyncUploadCallback.RESULT.ABORTED);
+                        callback.onAbort(new AsyncUploadResult(identifier, file));
                     }
                     throw new DataStoreException("Could not upload " + key, e2);
                 }
@@ -643,11 +612,10 @@
                 Thread.currentThread().setContextClassLoader(contextClassLoader);
             }
         }
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("write [" + identifier + "] length [" + file.length()
-                + "], in async mode [" + asyncUpload + "] in ["
-                + (System.currentTimeMillis() - start) + "] ms.");
-        }
+        LOG.debug(
+            "write of [{}], length=[{}], in async mode [{}], in [{}]ms",
+            new Object[] { identifier, file.length(), asyncUpload,
+                (System.currentTimeMillis() - start) });
     }
 
     /**
@@ -694,8 +662,8 @@
             } catch (InterruptedException ie) {
 
             }
-            LOG.info("Renamed [" + count + "] keys, time taken ["
-                + ((System.currentTimeMillis() - startTime) / 1000) + "] sec");
+            LOG.info("Renamed [{}] keys, time taken [{}]sec", count,
+                ((System.currentTimeMillis() - startTime) / 1000));
             // Delete older keys.
             if (deleteList.size() > 0) {
                 DeleteObjectsRequest delObjsReq = new DeleteObjectsRequest(
@@ -706,9 +674,10 @@
                     delObjsReq.setKeys(Collections.unmodifiableList(deleteList.subList(
                         startIndex, endIndex)));
                     DeleteObjectsResult dobjs = s3service.deleteObjects(delObjsReq);
-                    LOG.info("Records[" + dobjs.getDeletedObjects().size()
-                        + "] deleted in datastore from index [" + startIndex
-                        + "] to [" + (endIndex - 1) + "]");
+                    LOG.info(
+                        "Records[{}] deleted in datastore from index [{}] to [{}]",
+                        new Object[] { dobjs.getDeletedObjects().size(),
+                            startIndex, (endIndex - 1) });
                     if (endIndex == size) {
                         break;
                     } else {
@@ -775,9 +744,7 @@
                 CopyObjectRequest copReq = new CopyObjectRequest(bucket,
                     oldKey, bucket, newS3Key);
                 s3service.copyObject(copReq);
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug(oldKey + " renamed to " + newS3Key);
-                }
+                LOG.debug("[{}] renamed to [{}] ", oldKey, newS3Key);
             } finally {
                 if (contextClassLoader != null) {
                     Thread.currentThread().setContextClassLoader(
@@ -801,24 +768,34 @@
         private DataIdentifier identifier;
 
         private AsyncUploadCallback callback;
+        
+        private Upload upload;
 
-        public S3UploadProgressListener(DataIdentifier identifier, File file,
+        public S3UploadProgressListener(Upload upload, DataIdentifier identifier, File file,
                 AsyncUploadCallback callback) {
             super();
             this.identifier = identifier;
             this.file = file;
             this.callback = callback;
+            this.upload = upload;
         }
 
         public void progressChanged(ProgressEvent progressEvent) {
             switch (progressEvent.getEventCode()) {
                 case ProgressEvent.COMPLETED_EVENT_CODE:
-                    callback.call(identifier, file,
-                        AsyncUploadCallback.RESULT.SUCCESS);
+                    callback.onSuccess(new AsyncUploadResult(identifier, file));
                     break;
                 case ProgressEvent.FAILED_EVENT_CODE:
-                    callback.call(identifier, file,
-                        AsyncUploadCallback.RESULT.FAILED);
+                    AsyncUploadResult result = new AsyncUploadResult(
+                        identifier, file);
+                    try {
+                        AmazonClientException e = upload.waitForException();
+                        if (e != null) {
+                            result.setException(e);
+                        }
+                    } catch (InterruptedException e) {
+                    }
+                    callback.onFailure(result);
                     break;
                 default:
                     break;
Index: jackrabbit-aws-ext/src/main/java/org/apache/jackrabbit/aws/ext/S3Constants.java
===================================================================
--- jackrabbit-aws-ext/src/main/java/org/apache/jackrabbit/aws/ext/S3Constants.java	(revision 1579937)
+++ jackrabbit-aws-ext/src/main/java/org/apache/jackrabbit/aws/ext/S3Constants.java	(working copy)
@@ -31,6 +31,26 @@
      * Amazon aws secret key.
      */
     public static final String SECRET_KEY = "secretKey";
+    
+    /**
+     * Amazon S3 Http connection timeout.
+     */
+    public static final String S3_CONN_TIMEOUT = "connectionTimeout";
+    
+    /**
+     * Amazon S3  socket timeout.
+     */
+    public static final String S3_SOCK_TIMEOUT = "socketTimeout";
+    
+    /**
+     * Amazon S3  maximum connections to be used.
+     */
+    public static final String S3_MAX_CONNS = "maxConnections";
+    
+    /**
+     * Amazon S3  maximum retries.
+     */
+    public static final String S3_MAX_ERR_RETRY = "maxErrorRetry";
 
     /**
      * Amazon aws S3 bucket.
Index: jackrabbit-aws-ext/src/main/java/org/apache/jackrabbit/aws/ext/Utils.java
===================================================================
--- jackrabbit-aws-ext/src/main/java/org/apache/jackrabbit/aws/ext/Utils.java	(revision 1579937)
+++ jackrabbit-aws-ext/src/main/java/org/apache/jackrabbit/aws/ext/Utils.java	(working copy)
@@ -62,10 +62,10 @@
         AWSCredentials credentials = new BasicAWSCredentials(
             prop.getProperty(S3Constants.ACCESS_KEY),
             prop.getProperty(S3Constants.SECRET_KEY));
-        int connectionTimeOut = Integer.parseInt(prop.getProperty("connectionTimeout"));
-        int socketTimeOut = Integer.parseInt(prop.getProperty("socketTimeout"));
-        int maxConnections = Integer.parseInt(prop.getProperty("maxConnections"));
-        int maxErrorRetry = Integer.parseInt(prop.getProperty("maxErrorRetry"));
+        int connectionTimeOut = Integer.parseInt(prop.getProperty(S3Constants.S3_CONN_TIMEOUT));
+        int socketTimeOut = Integer.parseInt(prop.getProperty(S3Constants.S3_SOCK_TIMEOUT));
+        int maxConnections = Integer.parseInt(prop.getProperty(S3Constants.S3_MAX_CONNS));
+        int maxErrorRetry = Integer.parseInt(prop.getProperty(S3Constants.S3_MAX_ERR_RETRY));
         ClientConfiguration cc = new ClientConfiguration();
         cc.setConnectionTimeout(connectionTimeOut);
         cc.setSocketTimeout(socketTimeOut);
Index: jackrabbit-aws-ext/src/test/resources/log4j.properties
===================================================================
--- jackrabbit-aws-ext/src/test/resources/log4j.properties	(revision 1579937)
+++ jackrabbit-aws-ext/src/test/resources/log4j.properties	(working copy)
@@ -17,6 +17,8 @@
 
 # this is the log4j configuration for the JCR API tests
 log4j.rootLogger=INFO, file
+log4j.logger.org.apache.jackrabbit.core.data=DEBUG
+log4j.logger.org.apache.jackrabbit.aws.ext.ds=DEBUG
 
 #log4j.logger.org.apache.jackrabbit.test=DEBUG
 
Index: jackrabbit-aws-ext/src/test/resources/repository_sample.xml
===================================================================
--- jackrabbit-aws-ext/src/test/resources/repository_sample.xml	(revision 1579937)
+++ jackrabbit-aws-ext/src/test/resources/repository_sample.xml	(working copy)
@@ -43,6 +43,7 @@
         <param name="continueOnAsyncUploadFailure" value="false"/>
         <param name="concurrentUploadsThreads" value="10"/>
         <param name="asyncUploadLimit" value="100"/>
+        <param name="uploadRetries" value="3"/>
       </DataStore>
     <!--
         sample database data store configuration
Index: jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCache.java
===================================================================
--- jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCache.java	(revision 1579937)
+++ jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCache.java	(working copy)
@@ -93,14 +93,11 @@
             throws IOException {
         AsyncUploadCacheResult result = new AsyncUploadCacheResult();
         if (asyncUploadMap.entrySet().size() >= asyncUploadLimit) {
-            if (LOG.isInfoEnabled()) {
-                LOG.info("Async write limit [" + asyncUploadLimit
-                    + "] reached. File [" + fileName
-                    + "]  not added to async write cache.");
-            }
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("current set =" + asyncUploadMap.keySet());
-            }
+            LOG.info(
+                "Async write limit [{}]  reached. File [{}] not added to async write cache.",
+                asyncUploadLimit, fileName);
+            LOG.debug("AsyncUploadCache size=[{}] and entries =[{}]",
+                asyncUploadMap.size(), asyncUploadMap.keySet());
             result.setAsyncUpload(false);
         } else {
             long startTime = System.currentTimeMillis();
@@ -109,13 +106,10 @@
             }
             asyncUploadMap.put(fileName, System.currentTimeMillis());
             serializeAsyncUploadMap();
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("added file [" + fileName
-                    + "] to asyncUploadMap upoad took ["
-                    + ((System.currentTimeMillis() - startTime) / 1000)
-                    + "] sec");
-                LOG.debug("current set =" + asyncUploadMap.keySet());
-            }
+            LOG.debug("added file [{}] to asyncUploadMap upoad took [{}] sec",
+                fileName, ((System.currentTimeMillis() - startTime) / 1000));
+            LOG.debug("AsyncUploadCache size=[{}] and entries =[{}]",
+                asyncUploadMap.size(), asyncUploadMap.keySet());
             result.setAsyncUpload(true);
         }
         return result;
@@ -135,21 +129,15 @@
         Long retVal = asyncUploadMap.remove(fileName);
         if (retVal != null) {
             serializeAsyncUploadMap();
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("removed file [" + fileName
-                    + "] from asyncUploadMap took ["
-                    + ((System.currentTimeMillis() - startTime) / 1000)
-                    + "] sec");
-                LOG.debug("current set =" + asyncUploadMap.keySet());
-            }
+            LOG.debug("removed file [{}] from asyncUploadMap took [{}] sec",
+                fileName, ((System.currentTimeMillis() - startTime) / 1000));
+            LOG.debug("AsyncUploadCache size=[{}] and entries =[{}]",
+                asyncUploadMap.size(), asyncUploadMap.keySet());
         } else {
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("cannot remove file [" + fileName
-                    + "] from pending upoad took ["
-                    + ((System.currentTimeMillis() - startTime) / 1000)
-                    + "] sec. File not found");
-                LOG.debug("current set =" + asyncUploadMap.keySet());
-            }
+            LOG.debug("cannot removed file [{}] from asyncUploadMap took [{}] sec. File not found.",
+                fileName, ((System.currentTimeMillis() - startTime) / 1000));
+            LOG.debug("AsyncUploadCache size=[{}] and entries =[{}]",
+                asyncUploadMap.size(), asyncUploadMap.keySet());
         }
         AsyncUploadCacheResult result = new AsyncUploadCacheResult();
         result.setRequiresDelete(toBeDeleted.contains(fileName));
@@ -219,7 +207,7 @@
     public synchronized Set<String> deleteOlderThan(long min)
             throws IOException {
         min = min - 1000;
-        LOG.info("deleteOlderThan min =" + min);
+        LOG.info("deleteOlderThan min [{}]", min);
         Set<String> deleteSet = new HashSet<String>();
         for (Map.Entry<String, Long> entry : asyncUploadMap.entrySet()) {
             if (entry.getValue() < min) {
@@ -227,7 +215,7 @@
             }
         }
         if (deleteSet.size() > 0) {
-            LOG.info("deleteOlderThan set =" + deleteSet);
+            LOG.info("deleteOlderThan set [{}]", deleteSet);
             toBeDeleted.addAll(deleteSet);
             serializeToBeDeleted();
         }
@@ -247,8 +235,9 @@
         this.homeDir = homeDir;
         this.path = path;
         this.asyncUploadLimit = asyncUploadLimit;
-        LOG.info("AsynWriteCache:homeDir [" + homeDir + "], path [" + path
-            + "], asyncUploadLimit [" + asyncUploadLimit + "].");
+        LOG.info(
+            "AsynWriteCache:homeDir=[{}], path=[{}], asyncUploadLimit=[{}].",
+            new Object[] { homeDir, path, asyncUploadLimit });
         pendingUploads = new File(homeDir + "/" + PENDIND_UPLOAD_FILE);
         if (pendingUploads.exists()) {
             deserializeAsyncUploadMap();
Index: jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCallback.java
===================================================================
--- jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCallback.java	(revision 1579937)
+++ jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCallback.java	(working copy)
@@ -17,27 +17,24 @@
 
 package org.apache.jackrabbit.core.data;
 
-import java.io.File;
-
 /**
- * This interface defines method which would be called along with status.
+ * This interface defines callback methods to reflect the status of asynchronous
+ * upload.
  */
 public interface AsyncUploadCallback {
-
-    public void call(DataIdentifier identifier, File file, RESULT result);
-
-    public enum RESULT {
-        /**
-         * Asynchronous upload has succeeded.
-         */
-        SUCCESS,
-        /**
-         * Asynchronous upload has failed.
-         */
-        FAILED,
-        /**
-         * Asynchronous upload has been aborted.
-         */
-        ABORTED
-    };
+    
+    /**
+     * Callback method for successful asynchronous upload.
+     */
+    public void onSuccess(AsyncUploadResult result);
+    
+    /**
+     * Callback method for failed asynchronous upload.
+     */
+    public void onFailure(AsyncUploadResult result);
+    
+    /**
+     * Callback method for aborted asynchronous upload.
+     */
+    public void onAbort(AsyncUploadResult result);
 }
Index: jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadResult.java
===================================================================
--- jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadResult.java	(revision 0)
+++ jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadResult.java	(working copy)
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.File;
+
+/**
+ * 
+ * The class holds the result of asynchronous upload to {@link Backend}
+ */
+public class AsyncUploadResult {
+    /**
+     * {@link DataIdentifier} on which asynchronous upload is initiated.
+     */
+    private DataIdentifier identifier;
+    
+    /**
+     * {@link File} which is asynchronously uploaded.
+     */
+    private File file;
+    
+    /**
+     * Any {@link Exception} which is raised in asynchronously upload.
+     */
+    private Exception exception;
+    
+    public AsyncUploadResult(DataIdentifier identifier, File file) {
+        super();
+        this.identifier = identifier;
+        this.file = file;
+    }
+
+    public DataIdentifier getIdentifier() {
+        return identifier;
+    }
+
+    public File getFile() {
+        return file;
+    }
+
+    public Exception getException() {
+        return exception;
+    }
+
+    public void setIdentifier(DataIdentifier identifier) {
+        this.identifier = identifier;
+    }
+
+    public void setFile(File file) {
+        this.file = file;
+    }
+
+    public void setException(Exception exception) {
+        this.exception = exception;
+    }
+}
Index: jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/CachingDataStore.java
===================================================================
--- jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/CachingDataStore.java	(revision 1579937)
+++ jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/CachingDataStore.java	(working copy)
@@ -35,6 +35,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.WeakHashMap;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -73,6 +74,7 @@
  *     &lt;param name="{@link #setContinueOnAsyncUploadFailure(boolean) continueOnAsyncUploadFailure}" value="false"/>
  *     &lt;param name="{@link #setConcurrentUploadsThreads(int) concurrentUploadsThreads}" value="10"/>
  *     &lt;param name="{@link #setAsyncUploadLimit(int) asyncUploadLimit}" value="100"/>
+ *     &lt;param name="{@link #setUploadRetries(int) uploadRetries}" value="3"/>
  * &lt/DataStore>
  */
 public abstract class CachingDataStore extends AbstractDataStore implements
@@ -101,6 +103,15 @@
      * are garbage collected.
      */
     protected Map<DataIdentifier, WeakReference<DataIdentifier>> inUse = Collections.synchronizedMap(new WeakHashMap<DataIdentifier, WeakReference<DataIdentifier>>());
+    
+    /**
+     * In memory map to hold failed asynchronous upload {@link DataIdentifier}
+     * and its retry count. Once if all retries are exhausted or file is 
+     * successfully uploaded, then corresponding entry is flushed from the map.
+     * As all failed uploads are synchronously uploaded at startup, this map 
+     * is not required to be persisted. 
+     */
+    protected final Map<DataIdentifier, Integer> uploadRetryMap = new ConcurrentHashMap<DataIdentifier, Integer>(5);
 
     protected Backend backend;
 
@@ -145,6 +156,11 @@
      * The number of bytes in the cache. The default value is 64 GB.
      */
     private long cacheSize = 64L * 1024 * 1024 * 1024;
+    
+    /**
+     * The number of retries for failed upload.
+     */
+    private int uploadRetries = 3;
 
     /**
      * The local file system cache.
@@ -209,12 +225,12 @@
                 tmpDir = new File(path, "/repository/s3tmp");
                 path = path + "/repository/datastore";
             }
-            LOG.info("path=[" + path + ",] tmpPath= [" + tmpDir.getPath() + "]");
+            LOG.info("path=[{}],  tmpPath=[{}]", path, tmpDir.getPath());
             directory = new File(path);
             mkdirs(directory);
             if (!mkdirs(tmpDir)) {
                 FileUtils.cleanDirectory(tmpDir);
-                LOG.info("tmp = " + tmpDir.getPath() + " cleaned");
+                LOG.info("tmp=[{}] cleaned.", tmpDir.getPath());
             }
 
             asyncWriteCache = new AsyncUploadCache();
@@ -238,24 +254,25 @@
                                 + markerFile.getAbsolutePath(), e);
                     }
                 } else {
-                    LOG.info("marker file = " + markerFile.getAbsolutePath()
-                        + " exists");
+                    LOG.info("marker file = [{}] exists ",
+                        markerFile.getAbsolutePath());
                 }
             }
             // upload any leftover async uploads to backend during last shutdown
             Set<String> fileList = asyncWriteCache.getAll();
             if (fileList != null && !fileList.isEmpty()) {
                 List<String> errorFiles = new ArrayList<String>();
-                LOG.info("Uploading [" + fileList + "]  and size ["
-                    + fileList.size() + "] from AsyncUploadCache.");
+                LOG.info("Uploading [{}] and size=[{}] from AsyncUploadCache.",
+                    fileList, fileList.size());
                 long totalSize = 0;
                 List<File> files = new ArrayList<File>(fileList.size());
                 for (String fileName : fileList) {
                     File f = new File(path, fileName);
                     if (!f.exists()) {
                         errorFiles.add(fileName);
-                        LOG.error("Cannot upload pending file ["
-                            + f.getAbsolutePath() + "]. File doesn't exist.");
+                        LOG.error(
+                            "Cannot upload pending file [{}]. File doesn't exist.",
+                            f.getAbsolutePath());
                     } else {
                         totalSize += f.length();
                         files.add(new File(homeDir, fileName));
@@ -264,18 +281,22 @@
                 new FilesUploader(files, totalSize, concurrentUploadsThreads,
                     true).upload();
                 if (!continueOnAsyncUploadFailure && errorFiles.size() > 0) {
-                    LOG.error("Pending uploads of files [" + errorFiles
-                        + "] failed. Files do not exist in Local cache.");
-                    LOG.error("To continue set [continueOnAsyncUploadFailure] to true in Datastore configuration in repository.xml."
-                        + " There would be inconsistent data in repository due the missing files. ");
+                    LOG.error(
+                        "Pending uploads of files [{}] failed. Files do not exist in Local cache.",
+                        errorFiles);
+                    LOG.error("To continue set [continueOnAsyncUploadFailure] "
+                        + "to true in Datastore configuration in "
+                        + "repository.xml. There would be inconsistent data "
+                        + "in repository due the missing files. ");
                     throw new RepositoryException(
                         "Cannot upload async uploads from local cache. Files not found.");
                 } else {
                     if (errorFiles.size() > 0) {
-                        LOG.error("Pending uploads of files ["
-                            + errorFiles
-                            + "] failed. Files do not exist"
-                            + " in Local cache. Continuing as [continueOnAsyncUploadFailure] is set to true.");
+                        LOG.error(
+                            "Pending uploads of files [{}] failed. Files do" +
+                            " not exist in Local cache. Continuing as " +
+                            "[continueOnAsyncUploadFailure] is set to true.",
+                            errorFiles);
                     }
                     LOG.info("Reseting AsyncWrite Cache list.");
                     asyncWriteCache.reset();
@@ -324,10 +345,8 @@
             long currTime = System.currentTimeMillis();
             DataIdentifier identifier = new DataIdentifier(
                 encodeHexString(digest.digest()));
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("getting SHA1 hash  [" + identifier + "] length ["
-                    + length + "],   in [" + (currTime - startTime) + "] ms.");
-            }
+            LOG.debug("SHA1 of [{}], length =[{}] took [{}]ms ",
+                new Object[] { identifier, length, (currTime - startTime) });
             String fileName = getFileName(identifier);
             AsyncUploadCacheResult result = null;
             synchronized (this) {
@@ -337,10 +356,8 @@
                     result = cache.store(fileName, temporary, true);
                 }
             }
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("storing  [" + identifier + "] in localCache took ["
-                    + (System.currentTimeMillis() - currTime) + "] ms.");
-            }
+            LOG.debug("storing  [{}] in localCache took [{}] ms", identifier,
+                (System.currentTimeMillis() - currTime));
             if (result != null) {
                 if (result.canAsyncUpload()) {
                     backend.writeAsync(identifier, result.getFile(), this);
@@ -351,11 +368,9 @@
             // this will also make sure that
             // tempId is not garbage collected until here
             inUse.remove(tempId);
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("write [" + identifier + "] length [" + length
-                    + "],   in [" + (System.currentTimeMillis() - startTime)
-                    + "] ms.");
-            }
+            LOG.debug("addRecord [{}] of length [{}] took [{}]ms.",
+                new Object[] { identifier, length,
+                    (System.currentTimeMillis() - startTime) });
             return new CachingDataRecord(this, identifier);
         } catch (NoSuchAlgorithmException e) {
             throw new DataStoreException(DIGEST + " not available", e);
@@ -427,7 +442,7 @@
 
     @Override
     public void updateModifiedDateOnAccess(long before) {
-        LOG.info("minModifiedDate set to: " + before);
+        LOG.info("minModifiedDate set to [{}]", before);
         minModifiedDate = before;
     }
 
@@ -483,8 +498,9 @@
         } catch (IOException e) {
             throw new DataStoreException(e);
         }
-        LOG.info("deleteAllOlderThan  exit. Deleted [" + diSet
-            + "] records. Number of records deleted [" + diSet.size() + "]");
+        LOG.info(
+            "deleteAllOlderThan  exit. Deleted [{}]records. Number of records deleted [{}]",
+            diSet, diSet.size());
         return diSet.size();
     }
 
@@ -517,21 +533,19 @@
      */
     public long getLastModified(DataIdentifier identifier)
             throws DataStoreException {
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("accessed lastModified of identifier:" + identifier);
-        }
         String fileName = getFileName(identifier);
         long lastModified = asyncWriteCache.getLastModified(fileName);
         if (lastModified != 0) {
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("getlastModified of identifier [" + identifier
-                    + "] from AsyncUploadCache = " + lastModified);
-            }
-            return lastModified;
+            LOG.debug(
+                "identifier [{}]'s lastModified retrireved from AsyncUploadCache ",
+                identifier);
 
         } else {
-            return backend.getLastModified(identifier);
+            lastModified =  backend.getLastModified(identifier);
         }
+        LOG.debug("identifier= [{}], lastModified=[{}]", identifier,
+            lastModified);
+        return lastModified;
     }
 
     /**
@@ -575,49 +589,93 @@
     public Set<String> getPendingUploads() {
         return asyncWriteCache.getAll();
     }
+    
+    @Override
+    public void onSuccess(AsyncUploadResult result) {
+        DataIdentifier identifier = result.getIdentifier();
+        File file = result.getFile();
+        String fileName = getFileName(identifier);
+        try {
+            LOG.debug("Upload completed for [{}]", identifier);
+            // remove from failed upload map if any.
+            uploadRetryMap.remove(identifier);
+            AsyncUploadCacheResult cachedResult = asyncWriteCache.remove(fileName);
+            if (cachedResult.doRequiresDelete()) {
+                // added record already marked for delete
+                deleteRecord(identifier);
+            }
+        } catch (IOException ie) {
+            LOG.warn("Cannot remove pending file upload. Dataidentifer [ "
+                + identifier + "], file [" + file.getAbsolutePath() + "]", ie);
+        } catch (DataStoreException dse) {
+            LOG.warn("Cannot remove pending file upload. Dataidentifer [ "
+                + identifier + "], file [" + file.getAbsolutePath() + "]", dse);
+        }
+    }
 
-    public void call(DataIdentifier identifier, File file,
-            AsyncUploadCallback.RESULT resultCode) {
+    @Override
+    public void onFailure(AsyncUploadResult result) {
+        DataIdentifier identifier = result.getIdentifier();
+        File file = result.getFile();
         String fileName = getFileName(identifier);
-        if (AsyncUploadCallback.RESULT.SUCCESS.equals(resultCode)) {
-            try {
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug("Upload completed. [" + identifier + "].");
+        if (result.getException() != null) {
+            LOG.warn("Async Upload failed. Dataidentifer [ " + identifier
+                + "], file [" + file.getAbsolutePath() + "]",
+                result.getException());
+        } else {
+            LOG.warn("Async Upload failed. Dataidentifer [ " + identifier
+                + "], file [" + file.getAbsolutePath() + "]");
+        }
+        // Retry failed upload upto uploadRetries times.
+        try {
+            if (asyncWriteCache.hasEntry(fileName, false)) {
+                synchronized (uploadRetryMap) {
+                    Integer retry = uploadRetryMap.get(identifier);
+                    if (retry == null) {
+                        retry = new Integer(1);
+                    } else {
+                        retry++;
+                    }
+                    if (retry <= uploadRetries) {
+                        uploadRetryMap.put(identifier, retry);
+                        LOG.info(
+                            "Retring [{}] times failed upload for dataidentifer",
+                            retry, identifier);
+                        try {
+                            backend.writeAsync(identifier, file, this);
+                        } catch (DataStoreException e) {
+                            LOG.warn("exception", e);
+                        }
+                    } else {
+                        LOG.info("Retries [{}] exhausted for  dataidentifer.",
+                            (retry - 1), identifier);
+                        uploadRetryMap.remove(identifier);
+                    }
                 }
-                AsyncUploadCacheResult result = asyncWriteCache.remove(fileName);
-                if (result.doRequiresDelete()) {
-                    // added record already marked for delete
-                    deleteRecord(identifier);
-                }
-            } catch (IOException ie) {
-                LOG.warn("Cannot remove pending file upload. Dataidentifer [ "
-                    + identifier + "], file [" + file.getAbsolutePath() + "]",
-                    ie);
-            } catch (DataStoreException dse) {
-                LOG.warn("Cannot remove pending file upload. Dataidentifer [ "
-                    + identifier + "], file [" + file.getAbsolutePath() + "]",
-                    dse);
             }
-        } else if (AsyncUploadCallback.RESULT.FAILED.equals(resultCode)) {
-            LOG.error("Async Upload failed. Dataidentifer [ " + identifier
-                + "], file [" + file.getAbsolutePath() + "]");
-        } else if (AsyncUploadCallback.RESULT.ABORTED.equals(resultCode)) {
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Async Upload Aborted. Dataidentifer [ " + identifier
-                    + "], file [" + file.getAbsolutePath() + "]");
-            }
-            try {
-                asyncWriteCache.remove(fileName);
-                LOG.info("Async Upload Aborted. Dataidentifer [ " + identifier
-                    + "], file [" + file.getAbsolutePath() + "] removed.");
-            } catch (IOException ie) {
-                LOG.warn("Cannot remove pending file upload. Dataidentifer [ "
-                    + identifier + "], file [" + file.getAbsolutePath() + "]",
-                    ie);
-            }
+        } catch (IOException ie) {
+            LOG.warn("Cannot retry failed async file upload. Dataidentifer [ "
+                + identifier + "], file [" + file.getAbsolutePath() + "]", ie);
         }
     }
 
+    @Override
+    public void onAbort(AsyncUploadResult result) {
+        DataIdentifier identifier = result.getIdentifier();
+        File file = result.getFile();
+        String fileName = getFileName(identifier);
+        try {
+            asyncWriteCache.remove(fileName);
+            LOG.info(
+                "Async Upload Aborted. Dataidentifer [{}], file [{}] removed from AsyncCache.",
+                identifier, file.getAbsolutePath());
+        } catch (IOException ie) {
+            LOG.warn("Cannot remove pending file upload. Dataidentifer [ "
+                + identifier + "], file [" + file.getAbsolutePath() + "]", ie);
+        }
+    }
+
+
     /**
      * Returns a unique temporary file to be used for creating a new data
      * record.
@@ -636,44 +694,45 @@
         for (File f : files) {
             totalSize += f.length();
         }
-        if (concurrentUploadsThreads > 1) {
-            new FilesUploader(files, totalSize, concurrentUploadsThreads, false).upload();
-        } else {
-            uploadFilesInSingleThread(files, totalSize);
+        if (files.size() > 0) {
+            if (concurrentUploadsThreads > 1) {
+                new FilesUploader(files, totalSize, concurrentUploadsThreads,
+                    false).upload();
+            } else {
+                uploadFilesInSingleThread(files, totalSize);
+            }
         }
     }
 
     private void uploadFilesInSingleThread(List<File> files, long totalSize)
             throws RepositoryException {
         long startTime = System.currentTimeMillis();
-        LOG.info("Upload:  {" + files.size() + "} files in single thread.");
+        LOG.info("Upload:  [{}] files in single thread.", files.size());
         long currentCount = 0;
         long currentSize = 0;
         long time = System.currentTimeMillis();
         for (File f : files) {
-            long now = System.currentTimeMillis();
-            if (now > time + 5000) {
-                LOG.info("Uploaded:  {" + currentCount + "}/{" + files.size()
-                    + "} files, {" + currentSize + "}/{" + totalSize
-                    + "} size data");
-                time = now;
-            }
             String name = f.getName();
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("upload file = " + name);
-            }
+            LOG.debug("upload file [{}] ", name);
             if (!name.startsWith(TMP) && !name.endsWith(DS_STORE)
                 && f.length() > 0) {
                 uploadFileToBackEnd(f, false);
             }
             currentSize += f.length();
             currentCount++;
+            long now = System.currentTimeMillis();
+            if (now > time + 5000) {
+                LOG.info("Uploaded:  [{}/{}] files, [{}/{}] size data",
+                    new Object[] { currentCount, files.size(), currentSize,
+                        totalSize });
+                time = now;
+            }
         }
         long endTime = System.currentTimeMillis();
-        LOG.info("Uploaded:  {" + currentCount + "}/{" + files.size()
-            + "} files, {" + currentSize + "}/{" + totalSize
-            + "} size data, time taken {" + ((endTime - startTime) / 1000)
-            + "} sec");
+        LOG.info(
+            "Uploaded:  [{}/{}] files, [{}/{}] size data, time taken = [{}] sec",
+            new Object[] { currentCount, files.size(), currentSize, totalSize,
+                ((endTime - startTime) / 1000) });
     }
 
     /**
@@ -709,9 +768,7 @@
                 String fileName = getFileName(identifier);
                 asyncWriteCache.remove(fileName);
             }
-            if (LOG.isDebugEnabled()) {
-                LOG.debug(f.getName() + "uploaded.");
-            }
+            LOG.debug("uploaded [{}]", f.getName());
         } catch (IOException ioe) {
             throw new DataStoreException(ioe);
         }
@@ -910,7 +967,15 @@
             boolean continueOnAsyncUploadFailure) {
         this.continueOnAsyncUploadFailure = continueOnAsyncUploadFailure;
     }
+    
+    public int getUploadRetries() {
+        return uploadRetries;
+    }
 
+    public void setUploadRetries(int uploadRetries) {
+        this.uploadRetries = uploadRetries;
+    }
+
     public Backend getBackend() {
         return backend;
     }
@@ -962,15 +1027,14 @@
         }
 
         void logProgress() {
-            LOG.info("Uploaded:  {" + currentCount.get() + "}/{" + files.size()
-                + "} files, {" + currentSize.get() + "}/{" + totalSize
-                + "} size data");
+            LOG.info("Uploaded:  [{}/{}] files, [{}/{}] size data",
+                new Object[] { currentCount, files.size(), currentSize,
+                    totalSize });
         }
 
         void upload() throws DataStoreException {
             long startTime = System.currentTimeMillis();
-            LOG.info(" Uploading " + files.size() + " using " + threads
-                + " threads.");
+            LOG.info(" Uploading [{}] using [{}] threads.", files.size(), threads);
             ExecutorService executor = Executors.newFixedThreadPool(threads,
                 new NamedThreadFactory("backend-file-upload-worker"));
             int partitionSize = files.size() / (threads);
@@ -1005,10 +1069,10 @@
 
             }
             long endTime = System.currentTimeMillis();
-            LOG.info("Uploaded:  {" + currentCount.get() + "}/{" + files.size()
-                + "} files, {" + currentSize.get() + "}/{" + totalSize
-                + "} size data, time taken {" + ((endTime - startTime) / 1000)
-                + "} sec");
+            LOG.info(
+                "Uploaded:  [{}/{}] files, [{}/{}] size data, time taken = [{}] sec",
+                new Object[] { currentCount, files.size(), currentSize,
+                    totalSize, ((endTime - startTime) / 1000) });
             if (isExceptionRaised()) {
                 executor.shutdownNow(); // Cancel currently executing tasks
                 throw exception;
@@ -1044,12 +1108,10 @@
 
         public void run() {
             long time = System.currentTimeMillis();
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Thread [ " + Thread.currentThread().getName()
-                    + "]: Uploading files from startIndex[" + startIndex
-                    + "] and endIndex [" + (endIndex - 1)
-                    + "], both inclusive.");
-            }
+            LOG.debug(
+                "Thread [{}] : Uploading files from startIndex [{}] to endIndex [{}] both inclusive.",
+                new Object[] { Thread.currentThread().getName(), startIndex,
+                    (endIndex - 1) });
             int uploadCount = 0;
             long uploadSize = 0;
             try {
@@ -1059,9 +1121,7 @@
                         break;
                     }
                     String name = f.getName();
-                    if (LOG.isDebugEnabled()) {
-                        LOG.debug("upload file = " + name);
-                    }
+                    LOG.debug("upload file [{}] ",name);
                     if (!name.startsWith(TMP) && !name.endsWith(DS_STORE)
                         && f.length() > 0) {
                         uploadFileToBackEnd(f, updateAsyncCache);
Index: jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/LocalCache.java
===================================================================
--- jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/LocalCache.java	(revision 1579937)
+++ jackrabbit-data/src/main/java/org/apache/jackrabbit/core/data/LocalCache.java	(working copy)
@@ -109,9 +109,10 @@
         this.maxSize = size;
         directory = new File(path);
         tmp = new File(tmpPath);
-        LOG.info("cachePurgeTrigFactor = " + cachePurgeTrigFactor + ", cachePurgeResizeFactor = " + cachePurgeResizeFactor
-            + ", cachePurgeTrigFactorSize = " + (cachePurgeTrigFactor * size) + ", cachePurgeResizeFactor = "
-            + (cachePurgeResizeFactor * size));
+        LOG.info(
+            "cachePurgeTrigFactor =[{}], cachePurgeResizeFactor =[{}],  cachePurgeTrigFactorSize =[{}], cachePurgeResizeFactorSize =[{}]",
+            new Object[] { cachePurgeTrigFactor, cachePurgeResizeFactor,
+                (cachePurgeTrigFactor * size), (cachePurgeResizeFactor * size) });
         cache = new LRUCache(size, cachePurgeTrigFactor, cachePurgeResizeFactor);
         this.asyncUploadCache = asyncUploadCache;
 
@@ -151,13 +152,10 @@
                     && (f.getParentFile().exists() || f.getParentFile().mkdirs())
                     && transFile.renameTo(f) && f.exists()) {
                     if (transFile.exists() && transFile.delete()) {
-                        LOG.info("tmp file = " + transFile.getAbsolutePath()
-                            + " not deleted successfully");
+                        LOG.info("tmp file [{}] not deleted successfully", transFile.getAbsolutePath());
                     }
                     transFile = null;
-                    if (LOG.isDebugEnabled()) {
-                        LOG.debug("file [" + fileName + "] added to local cache.");
-                    }
+                    LOG.debug("file [{}] added to local cache.", fileName);
                     cache.put(fileName, f.length());
                 } else {
                     f = transFile;
@@ -223,9 +221,7 @@
                 dest.setLastModified(System.currentTimeMillis());
             }
             cache.put(fileName, dest.length());
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("file [" + fileName + "] added to local cache.");
-            }
+            LOG.debug("file [{}] added to local cache.", fileName);
             result.setFile(dest);
             if (tryForAsyncUpload) {
                 result.setAsyncUpload(asyncUploadCache.add(fileName).canAsyncUpload());
@@ -251,7 +247,9 @@
         // return file in purge mode = true and file present in asyncUploadCache
         // as asyncUploadCache's files will be not be deleted in cache purge.
         if (!f.exists() || (isInPurgeMode() && !asyncUploadCache.hasEntry(fileName, false))) {
-            log("purgeMode true or file doesn't exists: getFileIfStored returned");
+            LOG.debug(
+                "getFileIfStored returned: purgeMode=[{}], file=[{}] exists=[{}]",
+                new Object[] { isInPurgeMode(), fileName, f.exists() });
             return null;
         } else {
             // touch entry in LRU caches
@@ -269,7 +267,7 @@
      */
     public synchronized void delete(String fileName) {
         if (isInPurgeMode()) {
-            log("purgeMode true :delete returned");
+            LOG.debug("purgeMode true :delete returned");
             return;
         }
         fileName = fileName.replace("\\", "/");
@@ -302,7 +300,7 @@
      * unsuccessful delete files.
      */
     public void close() {
-        log("close");
+        LOG.debug("close");
         deleteOldFiles();
     }
 
@@ -315,7 +313,8 @@
       //order is important here
         boolean value = !isInPurgeMode() && (cache.canAdmitFile(length));
         if (!value) {
-            log("cannot admit file of length=" + length + " and currentSizeInBytes=" + cache.currentSizeInBytes);
+            LOG.debug("cannot admit file of length=[{}] and length=[{}] ",
+                length, cache.currentSizeInBytes);
         }
         return value;
     }
@@ -349,7 +348,7 @@
                 count++;
             }
         }
-        LOG.info("deleted [" + count + "]/[" + initialSize + "] files");
+        LOG.info("deleted [{}]/[{}] files.", count, initialSize);
     }
 
     /**
@@ -360,10 +359,10 @@
      * @return true if this method deletes file successfuly else return false.
      */
     boolean tryDelete(final String fileName) {
-        log("cache delete " + fileName);
+        LOG.debug("try deleting file [{}]", fileName);
         File f = getFile(fileName);
         if (f.exists() && f.delete()) {
-            log(fileName + "  deleted successfully");
+            LOG.debug("File [{}]  deleted successfully", fileName);
             toBeDeleted.remove(fileName);
             while (true) {
                 f = f.getParentFile();
@@ -375,7 +374,7 @@
             }
             return true;
         } else if (f.exists()) {
-            LOG.info("not able to delete file = " + f.getAbsolutePath());
+            LOG.info("not able to delete file [{}]", f.getAbsolutePath());
             toBeDeleted.add(fileName);
             return false;
         }
@@ -391,10 +390,6 @@
         return count;
     }
 
-    static void log(final String s) {
-        LOG.debug(s);
-    }
-
     /**
      * A LRU based extension {@link LinkedHashMap}. The key is file name and
      * value is length of file.
@@ -433,28 +428,30 @@
                 // not removing file from local cache, if there is in progress
                 // async upload on it.
                 if (asyncUploadCache.hasEntry(fileName, false)) {
-                    LOG.info("AsyncUploadCache upload contains file [" + fileName
-                        + "]. Not removing it from LocalCache.");
+                    LOG.info(
+                        "AsyncUploadCache upload contains file [{}]. Not removing it from LocalCache.",
+                        fileName);
                     return null;
                 }
             } catch (IOException e) {
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug("error: ", e);
-                }
+                LOG.debug("error: ", e);
                 return null;
             }
             Long flength = null;
             if (tryDelete(fileName)) {
                 flength = super.remove(key);
                 if (flength != null) {
-                    log("cache entry { " + fileName + "} with size {" + flength + "} removed.");
+                    LOG.debug("cache entry [{}], with size [{}] removed.",
+                        fileName, flength);
                     currentSizeInBytes -= flength.longValue();
                 }
             } else if (!getFile(fileName).exists()) {
                 // second attempt. remove from cache if file doesn't exists
                 flength = super.remove(key);
                 if (flength != null) {
-                    log(" file not exists. cache entry { " + fileName + "} with size {" + flength + "} removed.");
+                    LOG.debug(
+                        "file not exists. cache entry [{}], with size [{}] removed.",
+                        fileName, flength);
                     currentSizeInBytes -= flength.longValue();
                 }
             }
@@ -481,14 +478,15 @@
         synchronized void tryPurge() {
             if (currentSizeInBytes > cachePurgeTrigSize && !isInPurgeMode()) {
                 setPurgeMode(true);
-                LOG.info("currentSizeInBytes[" + cache.currentSizeInBytes + "] exceeds (cachePurgeTrigSize)[" + cache.cachePurgeTrigSize
-                    + "]");
+                LOG.info(
+                    "currentSizeInBytes=[{}]  exceeds cachePurgeTrigSize=[{}]",
+                    cache.currentSizeInBytes, cache.cachePurgeTrigSize);
                 new Thread(new PurgeJob()).start();
             } else {
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug("currentSizeInBytes[" + cache.currentSizeInBytes + "] and  (cachePurgeTrigSize)[" + cache.cachePurgeTrigSize
-                        + "], isInPurgeMode =[" + isInPurgeMode() + "]");
-                }
+                LOG.debug(
+                    "currentSizeInBytes=[{}],cachePurgeTrigSize=[{}], isInPurgeMode =[{}]",
+                    new Object[] { cache.currentSizeInBytes,
+                        cache.cachePurgeTrigSize, isInPurgeMode() });
             }
         }
         /**
@@ -537,10 +535,9 @@
                         }
 
                     }
-                    LOG.info(" cache purge job completed: cleaned ["
-                        + (initialSize - cache.size())
-                        + "] files and currentSizeInBytes = [ "
-                        + cache.currentSizeInBytes + "]");
+                    LOG.info(
+                        " cache purge job completed: cleaned [{}] files and currentSizeInBytes = [{}]",
+                        (initialSize - cache.size()), cache.currentSizeInBytes);
                 }
             } catch (Exception e) {
                 LOG.error("error in purge jobs:", e);
@@ -564,9 +561,8 @@
                 allFiles.add(f);
             }
             long t1 = System.currentTimeMillis();
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Time taken to recursive [" + allFiles.size() + "] took [" + ((t1 - startTime) / 1000) + "]sec");
-            }
+            LOG.debug("Time taken to recursive [{}] took [{}] sec",
+                allFiles.size(), ((t1 - startTime) / 1000));
             Collections.sort(allFiles, new Comparator<File>() {
                 public int compare(File o1, File o2) {
                     long l1 = o1.lastModified(), l2 = o2.lastModified();
@@ -574,9 +570,8 @@
                 }
             });
             long t2 = System.currentTimeMillis();
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Time taken to sort [" + allFiles.size() + "] took [" + ((t2 - t1) / 1000) + "]sec");
-            }
+            LOG.debug("Time taken to sort [{}] took [{}] sec",
+                allFiles.size(), ((t2 - t1) / 1000));
             String dataStorePath = directory.getAbsolutePath();
             long time = System.currentTimeMillis();
             int count = 0;
@@ -600,12 +595,14 @@
                     }
                 }
             }
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Processed {" + count + "}/{" + allFiles.size() + "} , currentSizeInBytes = " + cache.currentSizeInBytes
-                    + ",  maxSizeInBytes = " + cache.maxSizeInBytes + ",  cache.filecount = " + cache.size());
-            }
+            LOG.debug(
+                "Processed [{}]/[{}], currentSizeInBytes = [{}], maxSizeInBytes = [{}], cache.filecount = [{}]",
+                new Object[] { count, allFiles.size(),
+                    cache.currentSizeInBytes, cache.maxSizeInBytes,
+                    cache.size() });
             long t3 = System.currentTimeMillis();
-            LOG.info("Time to build cache of  [" + allFiles.size() + "] took [" + ((t3 - startTime) / 1000) + "]sec");
+            LOG.info("Time to build cache of  [{}] files took [{}] sec",
+                allFiles.size(), ((t3 - startTime) / 1000));
         }
     }
 }
Index: jackrabbit-data/src/test/java/org/apache/jackrabbit/core/data/InMemoryBackend.java
===================================================================
--- jackrabbit-data/src/test/java/org/apache/jackrabbit/core/data/InMemoryBackend.java	(revision 1579937)
+++ jackrabbit-data/src/test/java/org/apache/jackrabbit/core/data/InMemoryBackend.java	(working copy)
@@ -158,13 +158,12 @@
             timeMap.put(identifier, System.currentTimeMillis());
         } catch (IOException e) {
             if (async) {
-                callback.call(identifier, file,
-                    AsyncUploadCallback.RESULT.ABORTED);
+                callback.onAbort(new AsyncUploadResult(identifier, file));
             }
             throw new DataStoreException(e);
         }
         if (async) {
-            callback.call(identifier, file, AsyncUploadCallback.RESULT.SUCCESS);
+            callback.onSuccess(new AsyncUploadResult(identifier, file));
         }
     }