[PDI-10856] - Concat Fields Step Throws Array Index Out of Bounds Err…

…or When Row Level Logging is Enabled

build.xml

@@ -843,6 +843,7 @@
       <test todir="${testreports.xml.dir}" name="org.pentaho.di.cluster.MasterSlaveTest" haltonerror="false"/>
       <test todir="${testreports.xml.dir}" name="org.pentaho.di.cluster.PartitioningTest" haltonerror="false"/>
       <test todir="${testreports.xml.dir}" name="org.pentaho.di.cluster.SlaveSequenceTest" haltonerror="false"/>
+      <test todir="${testreports.xml.dir}" name="org.pentaho.di.trans.steps.concatfields.ConcatFields_PDI_10856_Test" haltonerror="false"/>
       <test todir="${testreports.xml.dir}" name="org.pentaho.di.www.CarteTest" haltonerror="false"/>
 
       <!-- Some extra things to run -->

engine/src/org/pentaho/di/trans/steps/concatfields/ConcatFields.java

@@ -145,9 +145,9 @@ public synchronized boolean processRow( StepMetaInterface smi, StepDataInterface
       // instead of writing to file, writes it to a stream
       writeRowToFile( data.inputRowMetaModified, r );
       setLinesOutput( 0 ); // we have to tweak it, no output here
-      putRowFromStream( r );
+      r = putRowFromStream( r );
     } else { // fast data dump
-      putRowFastDataDump( r );
+      r = putRowFastDataDump( r );
     }
 
     if ( log.isRowLevel() ) {
@@ -192,11 +192,11 @@ void prepareForReMap() throws KettleStepException {
   }
 
   // reads the row from the stream, flushs, add target field and call putRow()
-  void putRowFromStream( Object[] r ) throws KettleStepException {
+  Object[] putRowFromStream( Object[] r ) throws KettleStepException {
 
     byte[] targetBinary = ( (ConcatFieldsOutputStream) data.writer ).read();
     if ( r == null && targetBinary == null ) {
-      return; // special condition of header/footer/split
+      return null; // special condition of header/footer/split
     }
 
     Object[] outputRowData = prepareOutputRow( r );
@@ -221,11 +221,12 @@ void putRowFromStream( Object[] r ) throws KettleStepException {
     }
 
     putRow( data.outputRowMeta, outputRowData );
+    return outputRowData;
   }
 
   // concat as a fast data dump (no formatting) and call putRow()
   // this method is only called from a normal line, never from header/footer/split stuff
-  void putRowFastDataDump( Object[] r ) throws KettleStepException {
+  Object[] putRowFastDataDump( Object[] r ) throws KettleStepException {
 
     Object[] outputRowData = prepareOutputRow( r );
 
@@ -251,6 +252,7 @@ void putRowFastDataDump( Object[] r ) throws KettleStepException {
     outputRowData[data.posTargetField] = new String( targetString );
 
     putRow( data.outputRowMeta, outputRowData );
+    return outputRowData;
   }
 
   private void concatFieldFastDataDump( StringBuilder targetField, Object valueData, String nullString ) {

engine/test-src/org/pentaho/di/trans/steps/concatfields/ConcatFieldsTest.java

@@ -69,8 +69,8 @@ public void setRow( Object[] row ) {
     }
 
     @Override
-    protected void putRowFastDataDump( Object[] r ) throws KettleStepException {
-
+    protected Object[] putRowFastDataDump( Object[] r ) throws KettleStepException {
+      return null;
     }
 
     @Override
@@ -79,8 +79,8 @@ protected boolean writeHeader() {
     }
 
     @Override
-    void putRowFromStream( Object[] r ) throws KettleStepException {
-      prepareOutputRow( r );
+    Object[] putRowFromStream( Object[] r ) throws KettleStepException {
+      return prepareOutputRow( r );
     }
   }
 

test/org/pentaho/di/trans/steps/concatfields/ConcatFields_PDI_10856_Test.java

@@ -0,0 +1,50 @@
+/* ******************************************************************************
+ *
+ * Pentaho Data Integration
+ *
+ * Copyright (C) 2002-2015 by Pentaho : http://www.pentaho.com
+ *
+ *******************************************************************************
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ ******************************************************************************/
+
+package org.pentaho.di.trans.steps.concatfields;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.pentaho.di.core.KettleEnvironment;
+import org.pentaho.di.core.logging.LogLevel;
+import org.pentaho.di.trans.Trans;
+import org.pentaho.di.trans.TransMeta;
+
+public class ConcatFields_PDI_10856_Test {
+
+  @Test
+  public void rowLevelLoggingDoesNotFail() throws Exception {
+    KettleEnvironment.init();
+
+    TransMeta transMeta = new TransMeta( "testfiles/org/pentaho/di/trans/steps/concatfields/PDI-10856.ktr" );
+    transMeta.setTransformationType( TransMeta.TransformationType.Normal );
+
+    Trans trans = new Trans( transMeta );
+    trans.setLogLevel( LogLevel.ROWLEVEL );
+
+    trans.prepareExecution( null );
+    trans.startThreads();
+    trans.waitUntilFinished();
+
+    Assert.assertEquals(0, trans.getErrors());
+  }
+}