[BACKLOG-7997] - Can not execute partitioned clustered transformation…

… - fix NPE at clean-up and prepareExecution [BACKLOG-7997] - Can not execute partitioned clustered transformation - test
pentaho · May 4, 2016 · 6d5a6ba · 6d5a6ba
1 parent 962a58c
commit 6d5a6ba
Show file tree

Hide file tree

Showing 4 changed files with 100 additions and 10 deletions.
diff --git a/engine/src/org/pentaho/di/trans/Trans.java b/engine/src/org/pentaho/di/trans/Trans.java
@@ -3,7 +3,7 @@
  *
  * Pentaho Data Integration
  *
- * Copyright (C) 2002-2015 by Pentaho : http://www.pentaho.com
+ * Copyright (C) 2002-2016 by Pentaho : http://www.pentaho.com
  *
  *******************************************************************************
  *
@@ -1462,8 +1462,10 @@ protected void fireTransFinishedListeners() throws KettleException {
           badGuys.add( e );
         }
       }
-      // Signal for the the waitUntilFinished blocker...
-      transFinishedBlockingQueue.add( new Object() );
+      if ( transFinishedBlockingQueue != null ) {
+        // Signal for the the waitUntilFinished blocker...
+        transFinishedBlockingQueue.add( new Object() );
+      }
       if ( !badGuys.isEmpty() ) {
         // FIFO
         throw new KettleException( badGuys.get( 0 ) );

diff --git a/engine/src/org/pentaho/di/trans/step/BaseStep.java b/engine/src/org/pentaho/di/trans/step/BaseStep.java
@@ -3,7 +3,7 @@
  *
  * Pentaho Data Integration
  *
- * Copyright (C) 2002-2013 by Pentaho : http://www.pentaho.com
+ * Copyright (C) 2002-2016 by Pentaho : http://www.pentaho.com
  *
  *******************************************************************************
  *
@@ -781,10 +781,19 @@ public void cleanup() {
       }
     }
 
-    for ( RemoteStep remoteStep : getRemoteInputSteps() ) {
-      remoteStep.cleanup();
+    List<RemoteStep> remoteInputSteps = getRemoteInputSteps();
+    if ( remoteInputSteps != null ) {
+      cleanupRemoteSteps( remoteInputSteps );
+    }
+
+    List<RemoteStep> remoteOutputSteps = getRemoteOutputSteps();
+    if ( remoteOutputSteps != null ) {
+      cleanupRemoteSteps( remoteOutputSteps );
     }
-    for ( RemoteStep remoteStep : getRemoteOutputSteps() ) {
+  }
+
+  static void cleanupRemoteSteps( List<RemoteStep> remoteSteps ) {
+    for ( RemoteStep remoteStep : remoteSteps ) {
       remoteStep.cleanup();
     }
   }

diff --git a/engine/test-src/org/pentaho/di/trans/TransTest.java b/engine/test-src/org/pentaho/di/trans/TransTest.java
@@ -2,7 +2,7 @@
  *
  * Pentaho Data Integration
  *
- * Copyright (C) 2002-2015 by Pentaho : http://www.pentaho.com
+ * Copyright (C) 2002-2016 by Pentaho : http://www.pentaho.com
  *
  *******************************************************************************
  *
@@ -26,14 +26,19 @@
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -285,6 +290,27 @@ public void testRecordsCleanUpMethodIsCalled() throws Exception {
     verify( mockedDataBase ).cleanupLogRecords( stepLogTable );
   }
 
+  @Test
+  public void testFireTransFinishedListeners() throws Exception {
+    Trans trans = new Trans();
+    TransListener mockListener = mock( TransListener.class );
+    trans.setTransListeners( Collections.singletonList( mockListener ) );
+
+    trans.fireTransFinishedListeners();
+
+    verify( mockListener ).transFinished( trans );
+  }
+
+  @Test( expected = KettleException.class )
+  public void testFireTransFinishedListenersExceprionOnTransFinished() throws Exception {
+    Trans trans = new Trans();
+    TransListener mockListener = mock( TransListener.class );
+    doThrow( KettleException.class ).when( mockListener ).transFinished( trans );
+    trans.setTransListeners( Collections.singletonList( mockListener ) );
+
+    trans.fireTransFinishedListeners();
+  }
+
   private void startThreads( Runnable one, Runnable two, CountDownLatch start ) throws InterruptedException {
     Thread th = new Thread( one );
     Thread tt = new Thread( two );

diff --git a/engine/test-src/org/pentaho/di/trans/step/BaseStepTest.java b/engine/test-src/org/pentaho/di/trans/step/BaseStepTest.java
@@ -2,7 +2,7 @@
  *
  * Pentaho Data Integration
  *
- * Copyright (C) 2002-2015 by Pentaho : http://www.pentaho.com
+ * Copyright (C) 2002-2016 by Pentaho : http://www.pentaho.com
  *
  *******************************************************************************
  *
@@ -26,9 +26,12 @@
 import static org.mockito.Mockito.*;
 import static org.junit.Assert.*;
 
+import java.io.IOException;
+import java.net.ServerSocket;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Calendar;
+import java.util.Collections;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
@@ -57,6 +60,7 @@
 import org.pentaho.di.core.row.value.ValueMetaString;
 import org.pentaho.di.trans.BasePartitioner;
 import org.pentaho.di.trans.steps.mock.StepMockHelper;
+import org.pentaho.di.www.SocketRepository;
 
 public class BaseStepTest {
   private StepMockHelper<StepMetaInterface, StepDataInterface> mockHelper;
@@ -312,4 +316,53 @@ public void testBuildLog() throws KettleValueException {
     assertEquals( ValueMetaInterface.TYPE_DATE, result.getValueMeta( 8 ).getType() );
     assertEquals( endDate, result.getDate( 8, Calendar.getInstance().getTime() ) );
   }
+
+  @Test
+  public void testCleanupRemoteSteps() {
+    RemoteStep remoteStepMock = mock( RemoteStep.class );
+    BaseStep.cleanupRemoteSteps( Collections.singletonList( remoteStepMock ) );
+    verify( remoteStepMock ).cleanup();
+  }
+
+  @Test
+  public void testCleanup() throws IOException {
+    when( mockHelper.logChannelInterfaceFactory.create( any(), any( LoggingObjectInterface.class ) ) ).thenReturn(
+        mockHelper.logChannelInterface );
+    BaseStep baseStep =
+        new BaseStep( mockHelper.stepMeta, mockHelper.stepDataInterface, 0, mockHelper.transMeta, mockHelper.trans );
+    ServerSocket serverSocketMock = mock( ServerSocket.class );
+    doReturn( 0 ).when( serverSocketMock ).getLocalPort();
+    baseStep.setServerSockets( Collections.singletonList( serverSocketMock ) );
+    SocketRepository socketRepositoryMock = mock( SocketRepository.class );
+    baseStep.setSocketRepository( socketRepositoryMock );
+
+    baseStep.cleanup();
+
+    verify( socketRepositoryMock ).releaseSocket( 0 );
+  }
+
+  @Test
+  public void testCleanupWithInexistentRemoteSteps() throws IOException {
+    when( mockHelper.logChannelInterfaceFactory.create( any(), any( LoggingObjectInterface.class ) ) ).thenReturn(
+        mockHelper.logChannelInterface );
+    BaseStep baseStep =
+        spy( new BaseStep( mockHelper.stepMeta, mockHelper.stepDataInterface, 0, mockHelper.transMeta,
+            mockHelper.trans ) );
+    ServerSocket serverSocketMock = mock( ServerSocket.class );
+    doReturn( 0 ).when( serverSocketMock ).getLocalPort();
+    baseStep.setServerSockets( Collections.singletonList( serverSocketMock ) );
+    SocketRepository socketRepositoryMock = mock( SocketRepository.class );
+    baseStep.setSocketRepository( socketRepositoryMock );
+    RemoteStep inputStep = mock( RemoteStep.class );
+    doReturn( Collections.singletonList( inputStep ) ).when( baseStep ).getRemoteInputSteps();
+    RemoteStep outputStep = mock( RemoteStep.class );
+    doReturn( Collections.singletonList( outputStep ) ).when( baseStep ).getRemoteOutputSteps();
+
+    baseStep.cleanup();
+
+    verify( inputStep ).cleanup();
+    verify( outputStep ).cleanup();
+    verify( socketRepositoryMock ).releaseSocket( 0 );
+  }
+
 }