@Test public void testInitFailsOnNaNBatchAndDuration() { meta.setConsumerGroup( "testGroup" ); meta.setKafkaFactory( factory ); meta.setBatchDuration( "one" ); meta.setBatchSize( "two" ); step = new KafkaConsumerInput( stepMeta, data, 1, transMeta, trans ); assertFalse( step.init( meta, data ) ); verify( logChannel ).logError( "The \"Duration\" field is using a non-numeric value. Please set a numeric value." ); verify( logChannel ) .logError( "The \"Number of records\" field is using a non-numeric value. Please set a numeric value." ); }
@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 );  }