0000692: Escaped varchar data can get corrupted.

Also updated some csv logging.

symmetric/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/IncomingBatchService.java

@@ -39,6 +39,7 @@
 import org.jumpmind.symmetric.service.IIncomingBatchService;
 import org.jumpmind.symmetric.service.IParameterService;
 import org.jumpmind.symmetric.util.AppUtils;
+import org.jumpmind.util.FormatUtils;
 
 /**
  * @see IIncomingBatchService
@@ -200,7 +201,7 @@ public void insertIncomingBatch(IncomingBatch batch) {
                             batch.getFallbackInsertCount(), batch.getFallbackUpdateCount(),
                             batch.getIgnoreCount(), batch.getMissingDeleteCount(),
                             batch.getSkipCount(), batch.getSqlState(), batch.getSqlCode(),
-                            StringUtils.abbreviate(batch.getSqlMessage(), 1000),
+                            FormatUtils.abbreviateForLogging(batch.getSqlMessage()),
                             batch.getLastUpdatedHostName(), batch.getLastUpdatedTime() },
                     new int[] { Types.NUMERIC, Types.VARCHAR, Types.VARCHAR, Types.CHAR,
                             Types.NUMERIC, Types.NUMERIC, Types.NUMERIC, Types.NUMERIC,
@@ -237,7 +238,7 @@ public int updateIncomingBatch(IncomingBatch batch) {
                             batch.getFallbackUpdateCount(), batch.getIgnoreCount(),
                             batch.getMissingDeleteCount(), batch.getSkipCount(),
                             batch.getSqlState(), batch.getSqlCode(),
-                            StringUtils.abbreviate(batch.getSqlMessage(), 1000),
+                            FormatUtils.abbreviateForLogging(batch.getSqlMessage()),
                             batch.getLastUpdatedHostName(), batch.getLastUpdatedTime(),
                             batch.getBatchId(), batch.getNodeId() }, new int[] { Types.CHAR,
                             Types.SMALLINT, Types.NUMERIC, Types.NUMERIC, Types.NUMERIC,

symmetric/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/OutgoingBatchService.java

@@ -51,6 +51,7 @@
 import org.jumpmind.symmetric.service.IParameterService;
 import org.jumpmind.symmetric.service.ISequenceService;
 import org.jumpmind.symmetric.util.AppUtils;
+import org.jumpmind.util.FormatUtils;
 
 /**
  * @see IOutgoingBatchService
@@ -113,7 +114,7 @@ public void updateOutgoingBatch(OutgoingBatch outgoingBatch) {
                         outgoingBatch.getNetworkMillis(), outgoingBatch.getFilterMillis(),
                         outgoingBatch.getLoadMillis(), outgoingBatch.getExtractMillis(),
                         outgoingBatch.getSqlState(), outgoingBatch.getSqlCode(),
-                        StringUtils.abbreviate(outgoingBatch.getSqlMessage(), 1000),
+                        FormatUtils.abbreviateForLogging(outgoingBatch.getSqlMessage()),
                         outgoingBatch.getFailedDataId(), outgoingBatch.getLastUpdatedHostName(),
                         outgoingBatch.getLastUpdatedTime(), outgoingBatch.getBatchId(),
                         outgoingBatch.getNodeId() },

symmetric/symmetric-db/src/main/java/org/jumpmind/db/platform/AbstractDatabasePlatform.java

@@ -364,11 +364,7 @@ public Object[] getObjectValues(BinaryEncoding encoding, String[] values,
                         list.add(objectValue);
                     }
                 } catch (Exception ex) {
-                    String valueTrimmed = value;
-                    if (valueTrimmed.length() > 1000) {
-                        valueTrimmed = valueTrimmed.substring(0, 1000) + " ... (" + value.length()
-                                + " bytes)";
-                    }
+                    String valueTrimmed = FormatUtils.abbreviateForLogging(value);
                     log.error("Could not convert a value of {} for column {} of type {}",
                             new Object[] { valueTrimmed, column.getName(), column.getMappedType() });
                     log.error(ex.getMessage(), ex);

symmetric/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/CsvUtils.java

@@ -81,8 +81,12 @@ public static String escapeCsvData(String data) {
 
     public static String escapeCsvData(String[] data) {
         ByteArrayOutputStream out = new ByteArrayOutputStream();
+
         CsvWriter writer = new CsvWriter(new OutputStreamWriter(out), ',');
         writer.setEscapeMode(CsvWriter.ESCAPE_MODE_BACKSLASH);
+        writer.setTextQualifier('\"');
+        writer.setUseTextQualifier(true);
+        writer.setForceQualifier(true);
         for (String s : data) {
             try {
                 writer.write(s, true);
@@ -100,6 +104,8 @@ public static String escapeCsvData(String[] data, char recordDelimiter, char tex
         writer.setEscapeMode(CsvWriter.ESCAPE_MODE_BACKSLASH);        
         writer.setRecordDelimiter(recordDelimiter);
         writer.setTextQualifier(textQualifier);
+        writer.setUseTextQualifier(true);
+        writer.setForceQualifier(true);
         try {
             writer.writeRecord(data);
         } catch (IOException e) {

symmetric/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/reader/ProtocolDataReader.java

@@ -29,6 +29,7 @@
 import org.jumpmind.symmetric.io.data.IDataReader;
 import org.jumpmind.symmetric.io.stage.IStagedResource;
 import org.jumpmind.util.CollectionUtils;
+import org.jumpmind.util.FormatUtils;
 import org.jumpmind.util.Statistics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -123,10 +124,7 @@ protected Object readNext() {
                         bytesRead += token != null ? token.length() : 0;
                         if (debugBuffer != null) {
                             if (token != null) {
-                                String tokenTrimmed = token;
-                                if (tokenTrimmed.length() > 1000) {
-                                    tokenTrimmed = tokenTrimmed.substring(0, 1000) + " ... (" + token.length() + " bytes)";
-                                }
+                                String tokenTrimmed = FormatUtils.abbreviateForLogging(token);
                                 debugBuffer.append(tokenTrimmed);
                             } else {
                                 debugBuffer.append("<null>");

symmetric/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/AbstractProtocolDataWriter.java

@@ -16,8 +16,12 @@
 import org.jumpmind.symmetric.io.data.DataContext;
 import org.jumpmind.symmetric.io.data.IDataWriter;
 import org.jumpmind.util.Statistics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 abstract public class AbstractProtocolDataWriter implements IDataWriter {
+
+    protected final Logger log = LoggerFactory.getLogger(getClass());
 
     protected DataContext context;
 

symmetric/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/ProtocolDataWriter.java

@@ -7,11 +7,12 @@
 
 import org.jumpmind.exception.IoException;
 import org.jumpmind.symmetric.io.data.Batch;
+import org.jumpmind.util.FormatUtils;
 
 public class ProtocolDataWriter extends AbstractProtocolDataWriter {
 
     private BufferedWriter writer;
-
+    
     public ProtocolDataWriter(String sourceNodeId, Writer writer) {
         this(sourceNodeId, null, writer);
     }
@@ -41,6 +42,9 @@ protected void notifyEndBatch(Batch batch, IProtocolDataWriterListener listener)
     @Override
     protected void print(Batch batch, String data) {
         try {
+            if (log.isDebugEnabled() && data != null) {
+                log.debug("Writing data: {}", FormatUtils.abbreviateForLogging(data));
+            }
             writer.write(data);
         } catch (IOException e) {
             throw new IoException(e);

symmetric/symmetric-io/src/test/resources/org/jumpmind/symmetric/io/data/writer/FileCsvDataWriterTest.1.csv

@@ -7,9 +7,9 @@ schema,
 table,test
 keys,one
 columns,one,two,three
-insert,1,2,3
-old,1,2,3
-update,1,3,4,1
-old,1,2,3
-delete,1
+insert,"1","2","3"
+old,"1","2","3"
+update,"1","3","4","1"
+old,"1","2","3"
+delete,"1"
 commit,1

symmetric/symmetric-server/src/test/java/org/jumpmind/symmetric/test/SimpleIntegrationTest.java

@@ -176,14 +176,19 @@ public void syncToClient() {
         final String TEST_CLOB = "This is my test's test";
         // now change some data that should be sync'd
 
-        serverTestService.insertCustomer(new Customer(101, "Charlie Brown", true,
-                "300 Grub Street", "New Yorl", "NY", 90009, new Date(), new Date(), TEST_CLOB,
-                BIG_BINARY));
+        Customer customer = new Customer(101, "Charlie Brown", true,
+                "300 Grub Street \\o", "New Yorl", "NY", 90009, new Date(), new Date(), TEST_CLOB,
+                BIG_BINARY);
+        serverTestService.insertCustomer(customer);
 
         clientPull();
 
         Assert.assertTrue("The customer was not sync'd to the client."
                 + printRootAndClientDatabases(), clientTestService.doesCustomerExist(101));
+
+        Assert.assertEquals("The customer address was incorrect"
+                + printRootAndClientDatabases(), customer.getAddress(), clientTestService.getCustomer(101).getAddress());
+
 
         if (getServer().getSymmetricDialect().isClobSyncSupported()) {
             Assert.assertEquals("The CLOB notes field on customer was not sync'd to the client",

symmetric/symmetric-util/src/main/java/org/jumpmind/util/FormatUtils.java

@@ -11,9 +11,11 @@
 import org.apache.commons.lang.StringUtils;
 
 public final class FormatUtils {
-    
+
     public final static String WILDCARD = "*";
 
+    public final static int MAX_CHARS_TO_LOG = 1000;
+
     private static Pattern pattern = Pattern.compile("\\$\\((.+?)\\)");
 
     private FormatUtils() {
@@ -22,10 +24,11 @@ private FormatUtils() {
     public static String replace(String prop, String replaceWith, String sourceString) {
         return StringUtils.replace(sourceString, "$(" + prop + ")", replaceWith);
     }
-
-    public static String replaceToken(String text, String tokenToReplace, String replaceWithText, boolean matchUsingPrefixSuffix) {
+
+    public static String replaceToken(String text, String tokenToReplace, String replaceWithText,
+            boolean matchUsingPrefixSuffix) {
         Map<String, String> replacements = new HashMap<String, String>(1);
-        replacements.put(tokenToReplace,replaceWithText);
+        replacements.put(tokenToReplace, replaceWithText);
         return replaceTokens(text, replacements, matchUsingPrefixSuffix);
     }
 
@@ -80,7 +83,7 @@ public static String formatString(String format, String arg) {
             return String.format(format, arg);
         }
     }
-    
+
     public static boolean toBoolean(String value) {
         if (StringUtils.isNotBlank(value)) {
             if (value.equals("1")) {
@@ -94,18 +97,18 @@ public static boolean toBoolean(String value) {
             return false;
         }
     }
-    
+
     public static boolean isMixedCase(String text) {
         char[] chars = text.toCharArray();
         boolean upper = false;
         boolean lower = false;
         for (char ch : chars) {
             upper |= Character.isUpperCase(ch);
             lower |= Character.isLowerCase(ch);
-        }        
+        }
         return upper && lower;
     }
-    
+
     public static boolean isWildCardMatch(String text, String pattern) {
         boolean match = true;
         if (pattern.startsWith("^")) {
@@ -130,41 +133,42 @@ public static boolean isWildCardMatch(String text, String pattern) {
         }
 
         return match;
-    }   
-    
+    }
+
     /**
-     * Word wrap a string where the line size for the first line is different than 
-     * the lines sizes for the other lines.
+     * Word wrap a string where the line size for the first line is different
+     * than the lines sizes for the other lines.
      * 
      * @param str
      * @param firstLineSize
      * @param nonFirstLineSize
      * @return
      */
     public static String[] wordWrap(String str, int firstLineSize, int nonFirstLineSize) {
-        
+
         String[] lines = wordWrap(str, firstLineSize);
-        
+
         if (lines.length > 1 && firstLineSize != nonFirstLineSize) {
-            // More than one line.  Re-wrap the non-first lines with the non first line size
+            // More than one line. Re-wrap the non-first lines with the non
+            // first line size
             String notFirstLinesString = StringUtils.join(lines, " ", 1, lines.length);
             String[] nonFirstLines = wordWrap(notFirstLinesString, nonFirstLineSize);
             List<String> nonFirstLineCollection = Arrays.asList(nonFirstLines);
-            
+
             ArrayList<String> allLines = new ArrayList<String>();
             allLines.add(lines[0]);
             allLines.addAll(nonFirstLineCollection);
-            
+
             lines = allLines.toArray(lines);
         }
-        
+
         return lines;
     }
-
-
+
     public static String[] wordWrap(String str, int lineSize) {
         if (str != null && str.length() > lineSize) {
-            Pattern regex = Pattern.compile("(\\S\\S{" + lineSize + ",}|.{1," + lineSize + "})(\\s+|$)");
+            Pattern regex = Pattern.compile("(\\S\\S{" + lineSize + ",}|.{1," + lineSize
+                    + "})(\\s+|$)");
             List<String> list = new ArrayList<String>();
             Matcher m = regex.matcher(str);
             while (m.find()) {
@@ -177,13 +181,16 @@ public static String[] wordWrap(String str, int lineSize) {
                     // push line wrap and cause an unintentional blank line.
                     line = StringUtils.removeEnd(line, " ");
                     list.add(line);
-                }               
+                }
             }
             return (String[]) list.toArray(new String[list.size()]);
         } else {
             return new String[] { str };
         }
     }
-
+
+    public static String abbreviateForLogging(String value) {
+        return StringUtils.abbreviate(value, MAX_CHARS_TO_LOG);
+    }
 
 }