@Test public void testSendsEachRowToProducer_nullKey() throws Exception { TransMeta transMeta = new TransMeta( getClass().getResource( "/produceFourRows_nullKey.ktr" ).getPath() ); Trans trans = new Trans( transMeta ); trans.prepareExecution( new String[]{} ); StepMetaDataCombi combi = trans.getSteps().get( 1 ); KafkaProducerOutput step = (KafkaProducerOutput) combi.step; KafkaProducerOutputData data = (KafkaProducerOutputData) combi.data; when( kafkaFactory.producer( any(), any(), any(), any() ) ).thenReturn( kafkaProducer ); step.setKafkaFactory( kafkaFactory ); trans.startThreads(); trans.waitUntilFinished(); Mockito.verify( kafkaProducer, Mockito.times( 4 ) ).send( new ProducerRecord<>( "akastenka", "msg" ) ); Mockito.verify( kafkaProducer, Mockito.times( 1 ) ).close(); assertEquals( 4, trans.getSteps().get( 1 ).step.getLinesOutput() ); }
void setKafkaFactory( KafkaFactory factory ) { this.kafkaFactory = factory; }
KafkaProducerOutput extends BaseStep implements StepInterface { void setKafkaFactory( KafkaFactory factory ) { this.kafkaFactory = factory; } }
KafkaProducerOutput extends BaseStep implements StepInterface { void setKafkaFactory( KafkaFactory factory ) { this.kafkaFactory = factory; } KafkaProducerOutput( StepMeta stepMeta,
                              StepDataInterface stepDataInterface, int copyNr,
                              TransMeta transMeta, Trans trans ); }
KafkaProducerOutput extends BaseStep implements StepInterface { void setKafkaFactory( KafkaFactory factory ) { this.kafkaFactory = factory; } KafkaProducerOutput( StepMeta stepMeta,
                              StepDataInterface stepDataInterface, int copyNr,
                              TransMeta transMeta, Trans trans ); @Override boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); @Override void stopRunning( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ); }
KafkaProducerOutput extends BaseStep implements StepInterface { void setKafkaFactory( KafkaFactory factory ) { this.kafkaFactory = factory; } KafkaProducerOutput( StepMeta stepMeta,
                              StepDataInterface stepDataInterface, int copyNr,
                              TransMeta transMeta, Trans trans ); @Override boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); @Override void stopRunning( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface );  }