0001044: Map of semaphores used to prevent concurrent extraction of i…

…ndividual batches are not cleaned up
JumpMind · Feb 12, 2013 · d0a8d53 · d0a8d53
1 parent db39ac9
commit d0a8d53
Showing 1 changed file with 7 additions and 6 deletions.
diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java
@@ -117,6 +117,8 @@ public class DataExtractorService extends AbstractService implements IDataExtrac
     private IStatisticManager statisticManager;
 
     private IStagingManager stagingManager;
+
+    private Map<Long, Semaphore> locks = new HashMap<Long, Semaphore>();
 
     public DataExtractorService(IParameterService parameterService,
             ISymmetricDialect symmetricDialect, IOutgoingBatchService outgoingBatchService,
@@ -433,8 +435,6 @@ final protected OutgoingBatch requeryIfEnoughTimeHasPassed(long ts, OutgoingBatc
         return currentBatch;
     }
 
-    private Map<Long, Semaphore> locks = new HashMap<Long, Semaphore>();
-
     protected OutgoingBatch extractOutgoingBatch(Node targetNode,
             IDataWriter dataWriter, OutgoingBatch currentBatch,
             boolean streamToFileEnabled) {
@@ -477,14 +477,15 @@ protected OutgoingBatch extractOutgoingBatch(Node targetNode,
                     transformExtractWriter.close();
                 }
             } else {
-                if (!isPreviouslyExtracted(currentBatch)) {
+                if (currentBatch.getStatus() == Status.NE || 
+                        !isPreviouslyExtracted(currentBatch)) {
                     int maxPermits = parameterService.getInt(ParameterConstants.CONCURRENT_WORKERS);
                     Semaphore lock = null;
                     try {
                         synchronized (locks) {
                             lock = locks.get(currentBatch.getBatchId());
                             if (lock == null) {
-                                lock = new Semaphore(100);
+                                lock = new Semaphore(maxPermits);
                                 locks.put(currentBatch.getBatchId(), lock);
                             }
                             try {
@@ -523,7 +524,7 @@ protected OutgoingBatch extractOutgoingBatch(Node targetNode,
                         synchronized (locks) {
                             lock.release();
                             if (lock.availablePermits() == maxPermits) {
-                                locks.remove(lock);
+                                locks.remove(currentBatch.getBatchId());
                             }
                         }
                     }
@@ -1033,6 +1034,6 @@ public boolean containsData() {
             return data != null;
         }
 
-    }
+    }    
 
 }