@Test public void failingStart(TestContext context){ KafkaStreams streamsMock = mock(KafkaStreams.class); KStreamBuilder builderMock = mock(KStreamBuilder.class); Properties props = new Properties(); RuntimeVerticle verticleSpy = spy(new RuntimeVerticle(builderMock, props, null)); doThrow(RuntimeException.class).when(verticleSpy).createAndStartStream(); rule.vertx().deployVerticle(verticleSpy, context.asyncAssertFailure()); }
protected KafkaStreams createAndStartStream(){ streams = new KafkaStreams(builder, props); streams.setStateListener(((newState, oldState) -> { vertx.eventBus().publish(Config.CLUSTER_STATE_BROADCAST_ADDRESS, newState.toString()); LOG.info("State change in KafkaStreams recorded: oldstate=" + oldState + ", newstate=" + newState); if(listener != null) listener.onChange(newState, oldState); })); streams.start(); return streams; }
RuntimeVerticle extends AbstractVerticle { protected KafkaStreams createAndStartStream(){ streams = new KafkaStreams(builder, props); streams.setStateListener(((newState, oldState) -> { vertx.eventBus().publish(Config.CLUSTER_STATE_BROADCAST_ADDRESS, newState.toString()); LOG.info("State change in KafkaStreams recorded: oldstate=" + oldState + ", newstate=" + newState); if(listener != null) listener.onChange(newState, oldState); })); streams.start(); return streams; } }
RuntimeVerticle extends AbstractVerticle { protected KafkaStreams createAndStartStream(){ streams = new KafkaStreams(builder, props); streams.setStateListener(((newState, oldState) -> { vertx.eventBus().publish(Config.CLUSTER_STATE_BROADCAST_ADDRESS, newState.toString()); LOG.info("State change in KafkaStreams recorded: oldstate=" + oldState + ", newstate=" + newState); if(listener != null) listener.onChange(newState, oldState); })); streams.start(); return streams; } protected  RuntimeVerticle(KStreamBuilder builder, Properties props, KafkaStreams.StateListener listener); }
RuntimeVerticle extends AbstractVerticle { protected KafkaStreams createAndStartStream(){ streams = new KafkaStreams(builder, props); streams.setStateListener(((newState, oldState) -> { vertx.eventBus().publish(Config.CLUSTER_STATE_BROADCAST_ADDRESS, newState.toString()); LOG.info("State change in KafkaStreams recorded: oldstate=" + oldState + ", newstate=" + newState); if(listener != null) listener.onChange(newState, oldState); })); streams.start(); return streams; } protected  RuntimeVerticle(KStreamBuilder builder, Properties props, KafkaStreams.StateListener listener); @Override void start(Future<Void> startFuture); @Override void stop(Future<Void> stopFuture); }
RuntimeVerticle extends AbstractVerticle { protected KafkaStreams createAndStartStream(){ streams = new KafkaStreams(builder, props); streams.setStateListener(((newState, oldState) -> { vertx.eventBus().publish(Config.CLUSTER_STATE_BROADCAST_ADDRESS, newState.toString()); LOG.info("State change in KafkaStreams recorded: oldstate=" + oldState + ", newstate=" + newState); if(listener != null) listener.onChange(newState, oldState); })); streams.start(); return streams; } protected  RuntimeVerticle(KStreamBuilder builder, Properties props, KafkaStreams.StateListener listener); @Override void start(Future<Void> startFuture); @Override void stop(Future<Void> stopFuture);  }