@Test public void testProcessRow_first() throws Exception { meta.setConsumerGroup( "testGroup" ); meta.setKeyField( new KafkaConsumerField( KafkaConsumerField.Name.KEY, "key" ) ); meta.setMessageField( new KafkaConsumerField( KafkaConsumerField.Name.MESSAGE, "message" ) ); meta.setTopicField( new KafkaConsumerField( KafkaConsumerField.Name.TOPIC, null ) ); meta.setBatchDuration( "0" ); meta.setKafkaFactory( factory ); Collection<String> topics = Sets.newHashSet( topic.topic() ); step = spy( new KafkaConsumerInput( stepMeta, data, 1, transMeta, trans ) ); doReturn( null ).when( step ).getRow(); doNothing().when( step ).putRow( any( RowMetaInterface.class ), any( Object[].class ) ); when( factory.consumer( eq( meta ), any(), eq( meta.getKeyField().getOutputType() ), eq( meta.getMessageField().getOutputType() ) ) ).thenReturn( consumer ); int messageCount = 5; messages.put( topic, createRecords( topic.topic(), messageCount ) ); records = new ConsumerRecords<>( messages ); CountDownLatch latch = new CountDownLatch( 1 ); when( consumer.poll( anyLong() ) ).thenReturn( records ).then( invocationOnMock -> { latch.countDown(); return Collections.emptyList(); } ); topicList = new ArrayList<>(); topicList.add( topics.iterator().next() ); meta.setTopics( topicList ); step.init( meta, data ); KafkaStreamSource kafkaStreamSource = (KafkaStreamSource) spy( step.getSource() ); step.setSource( kafkaStreamSource ); List<Object> items = new ArrayList(); kafkaStreamSource.flowable().forEach( i -> items.add( i ) ); Runnable processRowRunnable = () -> { try { step.processRow( meta, data ); } catch ( KettleException e ) { fail( e.getMessage() ); } }; ExecutorService service = Executors.newSingleThreadExecutor(); service.submit( processRowRunnable ).get( 1, TimeUnit.SECONDS ); latch.await(); step.stopRunning( meta, data ); service.shutdown(); verify( kafkaStreamSource ).open(); verify( kafkaStreamSource, times( 2 ) ).flowable(); assertEquals( 5, items.size() ); assertNotNull( data.outputRowMeta.searchValueMeta( meta.getMessageField().getOutputName() ) ); assertNotNull( data.outputRowMeta.searchValueMeta( meta.getKeyField().getOutputName() ) ); assertNotNull( data.outputRowMeta.searchValueMeta( KafkaConsumerInputMeta.PARTITION_FIELD_NAME ) ); assertNotNull( data.outputRowMeta.searchValueMeta( KafkaConsumerInputMeta.OFFSET_FIELD_NAME ) ); assertNotNull( data.outputRowMeta.searchValueMeta( KafkaConsumerInputMeta.TIMESTAMP_FIELD_NAME ) ); assertNull( data.outputRowMeta.searchValueMeta( KafkaConsumerInputMeta.TOPIC_FIELD_NAME ) ); }
@Override public boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ) { KafkaConsumerInputMeta kafkaConsumerInputMeta = (KafkaConsumerInputMeta) stepMetaInterface; KafkaConsumerInputData kafkaConsumerInputData = (KafkaConsumerInputData) stepDataInterface; boolean superInit = super.init( kafkaConsumerInputMeta, kafkaConsumerInputData ); if ( !superInit ) { logError( BaseMessages.getString( PKG, "KafkaConsumerInput.Error.InitFailed" ) ); return false; } try { kafkaConsumerInputData.outputRowMeta = kafkaConsumerInputMeta.getRowMeta( getStepname(), this ); } catch ( KettleStepException e ) { log.logError( e.getMessage(), e ); } Consumer consumer = kafkaConsumerInputMeta.getKafkaFactory().consumer( kafkaConsumerInputMeta, this::environmentSubstitute, kafkaConsumerInputMeta.getKeyField().getOutputType(), kafkaConsumerInputMeta.getMessageField().getOutputType() ); Set<String> topics = kafkaConsumerInputMeta.getTopics().stream().map( this::environmentSubstitute ).collect( Collectors.toSet() ); consumer.subscribe( topics ); source = new KafkaStreamSource( consumer, kafkaConsumerInputMeta, kafkaConsumerInputData, variables, this ); window = new FixedTimeStreamWindow<>( getSubtransExecutor(), kafkaConsumerInputData.outputRowMeta, getDuration(), getBatchSize(), getParallelism(), kafkaConsumerInputMeta.isAutoCommit() ? p -> { } : this::commitOffsets ); return true; }
KafkaConsumerInput extends BaseStreamStep implements StepInterface { @Override public boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ) { KafkaConsumerInputMeta kafkaConsumerInputMeta = (KafkaConsumerInputMeta) stepMetaInterface; KafkaConsumerInputData kafkaConsumerInputData = (KafkaConsumerInputData) stepDataInterface; boolean superInit = super.init( kafkaConsumerInputMeta, kafkaConsumerInputData ); if ( !superInit ) { logError( BaseMessages.getString( PKG, "KafkaConsumerInput.Error.InitFailed" ) ); return false; } try { kafkaConsumerInputData.outputRowMeta = kafkaConsumerInputMeta.getRowMeta( getStepname(), this ); } catch ( KettleStepException e ) { log.logError( e.getMessage(), e ); } Consumer consumer = kafkaConsumerInputMeta.getKafkaFactory().consumer( kafkaConsumerInputMeta, this::environmentSubstitute, kafkaConsumerInputMeta.getKeyField().getOutputType(), kafkaConsumerInputMeta.getMessageField().getOutputType() ); Set<String> topics = kafkaConsumerInputMeta.getTopics().stream().map( this::environmentSubstitute ).collect( Collectors.toSet() ); consumer.subscribe( topics ); source = new KafkaStreamSource( consumer, kafkaConsumerInputMeta, kafkaConsumerInputData, variables, this ); window = new FixedTimeStreamWindow<>( getSubtransExecutor(), kafkaConsumerInputData.outputRowMeta, getDuration(), getBatchSize(), getParallelism(), kafkaConsumerInputMeta.isAutoCommit() ? p -> { } : this::commitOffsets ); return true; } }
KafkaConsumerInput extends BaseStreamStep implements StepInterface { @Override public boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ) { KafkaConsumerInputMeta kafkaConsumerInputMeta = (KafkaConsumerInputMeta) stepMetaInterface; KafkaConsumerInputData kafkaConsumerInputData = (KafkaConsumerInputData) stepDataInterface; boolean superInit = super.init( kafkaConsumerInputMeta, kafkaConsumerInputData ); if ( !superInit ) { logError( BaseMessages.getString( PKG, "KafkaConsumerInput.Error.InitFailed" ) ); return false; } try { kafkaConsumerInputData.outputRowMeta = kafkaConsumerInputMeta.getRowMeta( getStepname(), this ); } catch ( KettleStepException e ) { log.logError( e.getMessage(), e ); } Consumer consumer = kafkaConsumerInputMeta.getKafkaFactory().consumer( kafkaConsumerInputMeta, this::environmentSubstitute, kafkaConsumerInputMeta.getKeyField().getOutputType(), kafkaConsumerInputMeta.getMessageField().getOutputType() ); Set<String> topics = kafkaConsumerInputMeta.getTopics().stream().map( this::environmentSubstitute ).collect( Collectors.toSet() ); consumer.subscribe( topics ); source = new KafkaStreamSource( consumer, kafkaConsumerInputMeta, kafkaConsumerInputData, variables, this ); window = new FixedTimeStreamWindow<>( getSubtransExecutor(), kafkaConsumerInputData.outputRowMeta, getDuration(), getBatchSize(), getParallelism(), kafkaConsumerInputMeta.isAutoCommit() ? p -> { } : this::commitOffsets ); return true; } KafkaConsumerInput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                             Trans trans ); }
KafkaConsumerInput extends BaseStreamStep implements StepInterface { @Override public boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ) { KafkaConsumerInputMeta kafkaConsumerInputMeta = (KafkaConsumerInputMeta) stepMetaInterface; KafkaConsumerInputData kafkaConsumerInputData = (KafkaConsumerInputData) stepDataInterface; boolean superInit = super.init( kafkaConsumerInputMeta, kafkaConsumerInputData ); if ( !superInit ) { logError( BaseMessages.getString( PKG, "KafkaConsumerInput.Error.InitFailed" ) ); return false; } try { kafkaConsumerInputData.outputRowMeta = kafkaConsumerInputMeta.getRowMeta( getStepname(), this ); } catch ( KettleStepException e ) { log.logError( e.getMessage(), e ); } Consumer consumer = kafkaConsumerInputMeta.getKafkaFactory().consumer( kafkaConsumerInputMeta, this::environmentSubstitute, kafkaConsumerInputMeta.getKeyField().getOutputType(), kafkaConsumerInputMeta.getMessageField().getOutputType() ); Set<String> topics = kafkaConsumerInputMeta.getTopics().stream().map( this::environmentSubstitute ).collect( Collectors.toSet() ); consumer.subscribe( topics ); source = new KafkaStreamSource( consumer, kafkaConsumerInputMeta, kafkaConsumerInputData, variables, this ); window = new FixedTimeStreamWindow<>( getSubtransExecutor(), kafkaConsumerInputData.outputRowMeta, getDuration(), getBatchSize(), getParallelism(), kafkaConsumerInputMeta.isAutoCommit() ? p -> { } : this::commitOffsets ); return true; } KafkaConsumerInput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                             Trans trans ); @Override boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ); }
KafkaConsumerInput extends BaseStreamStep implements StepInterface { @Override public boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ) { KafkaConsumerInputMeta kafkaConsumerInputMeta = (KafkaConsumerInputMeta) stepMetaInterface; KafkaConsumerInputData kafkaConsumerInputData = (KafkaConsumerInputData) stepDataInterface; boolean superInit = super.init( kafkaConsumerInputMeta, kafkaConsumerInputData ); if ( !superInit ) { logError( BaseMessages.getString( PKG, "KafkaConsumerInput.Error.InitFailed" ) ); return false; } try { kafkaConsumerInputData.outputRowMeta = kafkaConsumerInputMeta.getRowMeta( getStepname(), this ); } catch ( KettleStepException e ) { log.logError( e.getMessage(), e ); } Consumer consumer = kafkaConsumerInputMeta.getKafkaFactory().consumer( kafkaConsumerInputMeta, this::environmentSubstitute, kafkaConsumerInputMeta.getKeyField().getOutputType(), kafkaConsumerInputMeta.getMessageField().getOutputType() ); Set<String> topics = kafkaConsumerInputMeta.getTopics().stream().map( this::environmentSubstitute ).collect( Collectors.toSet() ); consumer.subscribe( topics ); source = new KafkaStreamSource( consumer, kafkaConsumerInputMeta, kafkaConsumerInputData, variables, this ); window = new FixedTimeStreamWindow<>( getSubtransExecutor(), kafkaConsumerInputData.outputRowMeta, getDuration(), getBatchSize(), getParallelism(), kafkaConsumerInputMeta.isAutoCommit() ? p -> { } : this::commitOffsets ); return true; } KafkaConsumerInput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                             Trans trans ); @Override boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface );  }