0001438: Improve memory footprint usage when big lobs are in flight

JumpMind · Oct 11, 2013 · a4cae76 · a4cae76
1 parent b8adb92
commit a4cae76
Show file tree

Hide file tree

Showing 5 changed files with 22 additions and 9 deletions.
diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java
@@ -865,8 +865,8 @@ public void batchInError(DataContext context, Throwable ex) {
                             this.currentBatch.getNodeBatchId(), ex.getMessage());
                     this.currentBatch.setSqlMessage(ex.getMessage());
                 } else {
-                    log.error("Failed to load batch {} because: {}", new Object[] {
-                            this.currentBatch.getNodeBatchId(), ex.getMessage() });
+                    log.error(String.format("Failed to load batch %s because: %s", 
+                            this.currentBatch.getNodeBatchId(), ex.getMessage()), ex);                    
                     if (log.isDebugEnabled()) {
                         log.debug(ex.getMessage(), ex);
                     }

diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/CsvData.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/CsvData.java
@@ -146,6 +146,9 @@ public String getCsvData(String key) {
             String[] parsedData = parsedCsvData.get(key);
             if (parsedData != null) {
                 data = CsvUtils.escapeCsvData(parsedData);
+                // swap out data for parsed data so we don't 
+                // don't double the amount of memory being used
+                putCsvData(key, data);
             }
         }
         return data;

diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/reader/ExtractDataReader.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/reader/ExtractDataReader.java
@@ -40,8 +40,12 @@
 import org.jumpmind.symmetric.io.data.IDataReader;
 import org.jumpmind.util.CollectionUtils;
 import org.jumpmind.util.Statistics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class ExtractDataReader implements IDataReader {
+
+    protected static final Logger log = LoggerFactory.getLogger(ExtractDataReader.class);
 
     protected Map<Batch, Statistics> statistics = new HashMap<Batch, Statistics>();
 
@@ -175,6 +179,7 @@ protected CsvData enhanceWithLobsFromSourceIfNeeded(Table table, CsvData data) {
                             } else {
                                 valueForCsv = new String(binaryData);
                             }
+                            binaryData = null;
                         }
                     } else {
                         valueForCsv = sqlTemplate.queryForClob(sql, args);

diff --git a/...ic-io/src/main/java/org/jumpmind/symmetric/io/data/writer/AbstractProtocolDataWriter.java b/...ic-io/src/main/java/org/jumpmind/symmetric/io/data/writer/AbstractProtocolDataWriter.java
@@ -242,16 +242,17 @@ protected int println(String key, Column[] columns) {
     abstract protected void print(Batch batch, String data);
 
     protected long println(String... data) {
-        StringBuilder buffer = new StringBuilder();
+        long byteCount = 0;
         for (int i = 0; i < data.length; i++) {
             if (i != 0) {
-                buffer.append(delimiter);
+                print(batch, delimiter);
+                byteCount += delimiter.length();
             }
-            buffer.append(data[i]);
+            print(batch, data[i]);
+            byteCount += data[i].length();
         }
-        buffer.append("\n");
-        print(batch, buffer.toString());
-        long byteCount = buffer.length();
+        print(batch, "\n");
+        byteCount += "\n".length();
         statistics.get(batch).increment(DataWriterStatisticConstants.BYTECOUNT, byteCount);
         return byteCount;
     }

diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/StagingDataWriter.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/StagingDataWriter.java
@@ -108,7 +108,11 @@ protected void print(Batch batch, String data) {
         IStagedResource resource = getStagedResource(batch);
         BufferedWriter writer = resource.getWriter();
         try {
-            writer.append(data);
+            int size = data.length();
+            for (int i = 0; i < size; i = i + 1024) {
+                int end = i + 1024;
+                writer.append(data, i, end < size ? end : size);
+            }
         } catch (IOException ex) {
             throw new IoException(ex);
         }