TEIID-4744 more quickly calling close

teiid · Feb 3, 2017 · 46bbbd7 · 46bbbd7
1 parent cb827b1
commit 46bbbd7
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 1 deletion.
diff --git a/engine/src/main/java/org/teiid/dqp/internal/process/DataTierManagerImpl.java b/engine/src/main/java/org/teiid/dqp/internal/process/DataTierManagerImpl.java
@@ -170,7 +170,9 @@ private void signalMore() {
 
 		@Override
 		public void onCompletion(FutureWork<Void> future) {
-			signalMore();
+		    if (future != null) {
+		        signalMore();
+		    }
 			toRead.closeSource();
 			dtts.fullyCloseSource();
 		}
@@ -1118,13 +1120,18 @@ public List<?> nextTuple() throws TeiidComponentException,
 						TeiidProcessingException {
 					if (!processed) {
 						callable.call();
+						callable.onCompletion(null);
 						processed = true;
 					}
 					return ts.nextTuple();
 				}
 
 				@Override
 				public void closeSource() {
+				    if (!processed) {
+				        callable.onCompletion(null);
+				        processed = true;
+				    }
 					ts.closeSource();
 				}
 			};

diff --git a/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java b/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java
@@ -489,6 +489,21 @@ public RequestMessage exampleRequestMessage(String sql) {
     	helpTestSourceConcurrencyWithLimitedUnion();
     }
 
+    @Test public void testSerialThreadBoundClose() throws Exception {
+        agds.setSleep(100);
+        agds.threadBound = true;
+        core.setUserRequestSourceConcurrency(1);
+        String sql = "SELECT IntKey FROM BQT1.SmallA"; //$NON-NLS-1$
+        String userName = "logon"; //$NON-NLS-1$
+        ((BufferManagerImpl)core.getBufferManager()).setProcessorBatchSize(1);
+
+        RequestMessage reqMsg = exampleRequestMessage(sql);
+        //execute by don't finish the work
+        execute(userName, 1, reqMsg);
+        //make sure the source request is still closed
+        assertEquals(1, agds.getCloseCount().get());
+    }
+
     @Test(expected=TeiidProcessingException.class) public void testThreadBoundException() throws Exception {
     	agds.threadBound = true;
     	agds.throwExceptionOnExecute = true;