Merge branch '3.10' of https://github.com/JumpMind/symmetric-ds.git i…

…nto 3.10
JumpMind · Apr 12, 2019 · 89d17f9 · 89d17f9
2 parents e172551 + b7d38cb
commit 89d17f9
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 10 deletions.
diff --git a/symmetric-assemble/src/asciidoc/configuration/ldap.ad b/symmetric-assemble/src/asciidoc/configuration/ldap.ad
@@ -6,6 +6,11 @@ Configuring a <<Users>> authentication on SymmetricDS using LDAP is as simple as
 console.auth.ldap.baseDN::  The LDAP base DN to search for a user.
                             [&#xA0;Default:&#xA0;]
 
+
+A list of base DNs can be specified by separating each entry by a pipe ("|") symbol. +
+For example: +
+ou=Users,o=IT,c=US,dc=corp,dc=local|ou=Users,o=Tech,c=US,dc=corp,dc=local
+
 console.auth.ldap.host::  The LDAP server host name.
                             [&#xA0;Default:&#xA0;]
 

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
@@ -614,18 +614,19 @@ protected List<IncomingBatch> loadDataFromTransport(final ProcessInfo transferIn
 
                 OutputStreamWriter outWriter = null;
                 if (out != null) {
-                    outWriter = new OutputStreamWriter(out, IoConstants.ENCODING);
-                    long keepAliveMillis = parameterService.getLong(ParameterConstants.DATA_LOADER_SEND_ACK_KEEPALIVE);
-                    while (!executor.awaitTermination(keepAliveMillis, TimeUnit.MILLISECONDS)) {
-                        outWriter.write("1=1&");
-                        outWriter.flush();
+                    try {                        
+                        outWriter = new OutputStreamWriter(out, IoConstants.ENCODING);
+                        long keepAliveMillis = parameterService.getLong(ParameterConstants.DATA_LOADER_SEND_ACK_KEEPALIVE);
+                        while (!executor.awaitTermination(keepAliveMillis, TimeUnit.MILLISECONDS)) {
+                            outWriter.write("1=1&");
+                            outWriter.flush();
+                        }
+                    } catch (Exception ex) {
+                        log.warn("Failed to send keep alives to " + sourceNode + " " + ex.toString());
+                        awaitTermination(executor);          
                     }
                 } else {
-                    long hours = 1;
-                    while (!executor.awaitTermination(1, TimeUnit.HOURS)) {
-                        log.info(String.format("Executor has been awaiting loader termination for %d hour(s).", hours));
-                        hours++;
-                    }            
+                    awaitTermination(executor);            
                 }
 
                 loadListener.isDone();
@@ -679,6 +680,14 @@ protected IDataWriter chooseDataWriter(Batch batch) {
         return listener.getBatchesProcessed();
     }
 
+    private void awaitTermination(ExecutorService executor) throws InterruptedException {
+        long hours = 1;
+        while (!executor.awaitTermination(1, TimeUnit.HOURS)) {
+            log.info(String.format("Executor has been awaiting loader termination for %d hour(s).", hours));
+            hours++;
+        }
+    }
+
     protected void logOrRethrow(Throwable ex) throws IOException {
         // Throwing exception will mean acks are not sent, so only certain exceptions should be thrown
         if (ex instanceof RegistrationRequiredException) {