@Test public void kafkaClientClosedOnStop() throws Exception { TransMeta transMeta = new TransMeta( getClass().getResource( "/produceForever.ktr" ).getPath() ); Trans trans = new Trans( transMeta ); trans.prepareExecution( new String[]{} ); StepMetaDataCombi combi = trans.getSteps().get( 1 ); KafkaProducerOutput step = (KafkaProducerOutput) combi.step; when( kafkaFactory.producer( any(), any(), any(), any() ) ).thenReturn( kafkaProducer ); when( kafkaProducer.send( any() ) ).then( ignore -> { trans.stopAll(); return null; } ); step.setKafkaFactory( kafkaFactory ); trans.startThreads(); trans.waitUntilFinished(); verify( kafkaProducer ).flush(); verify( kafkaProducer ).close(); }
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 );  }