def METHOD_NAME(tmp_topic1, tmp_topic2):
    config = {
        "bootstrap.servers": KAFKA_BROKER,
    }
    topics = [tmp_topic1, tmp_topic2]
    producer = Producer(config)
    inp = []
    for i, topic in enumerate(topics):
        for j in range(3):
            key = f"key-{i}-{j}".encode()
            value = f"value-{i}-{j}".encode()
            producer.produce(topic, value, key)
            inp.append((key, value))
    producer.flush()
    flow = Dataflow()
    flow.input("inp", KafkaInput([KAFKA_BROKER], topics, tail=False))
    out = []
    flow.output("out", TestingOutput(out))
    run_main(flow)
    assert sorted(out) == sorted(inp)