def test__given__stream_already_started__when__call_start_again__then__raise_error(): kafka_config.NUM_STREAM_THREADS = 0 topology_builder = TopologyBuilder() topology_builder.source('my-source', ['my-input-topic-1']) topology_builder.processor('my-processor', MyTestProcessor, 'my-source') topology_builder.sink('my-sink', 'my-output-topic-1', 'my-processor') topology = topology_builder.build() kafka_streams = KafkaStreams(topology, kafka_config) kafka_streams.start() with pytest.raises(KafkaStreamsError, message='KafkaStreams already started.'): kafka_streams.start()