GnnForPrivacyScan/data/purposeSplit/kafka/kafka_1.py

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()