<p>I have my own Spring Cloud Data Flow processor with Python inside, I used this sample as a guide: <a href="https://dataflow.spring.io/docs/recipes/polyglot/processor/" rel="nofollow noreferrer">https://dataflow.spring.io/docs/recipes/polyglot/processor/</a>.
Then I want to scale and create three of these processors, so using <code>spring.cloud.deployer.myApp.count=3</code> I create 3 pods with Python inside.
I slightly modified a code in sample: when I create a Kafka consumer, I also pass a group id, so messages should be load balanced.</p>

<pre><code>consumer = KafkaConsumer(get_input_channel(), group_id=get_consumer_group(), bootstrap_servers=[get_kafka_binder_brokers()])
</code></pre>

<p>The issue is that SCDF creates a Kafka topic with only 1 partition, so messages arrive to one pod only. 
So I am wondering:</p>

<ul>
<li>Should I somehow configure SCDF to create a Kafka topic with 3 partitions?</li>
<li>Or should I not rely on SCDF and create topics on my own in Python? I suppose this will be redundant, since SCDF also creates this topic.</li>
<li>What component in SCDF is actually responsible for Kafka topics creation? And how can I influence it regarding number of partitions?</li>
<li>If I stop this stream and launch again with 4 processor steps, should the topic be extended with the 4th partition? Because currently no new partitions get created.</li>
</ul>

<p>Was able to get around this by introducing the following code into the Python container startup script, improved code provided in <a href="https://dataflow.spring.io/docs/recipes/polyglot/processor/" rel="nofollow noreferrer">https://dataflow.spring.io/docs/recipes/polyglot/processor/</a> . Using arguments passed by SCDF server to get a broker URL, topic name, number of instances:</p>

<pre><code>admin_client = KafkaAdminClient(bootstrap_servers=[get_kafka_binder_brokers()], client_id=sys.argv[0])

partition_count = get_cmd_arg("spring.cloud.stream.instanceCount")

# create Kafka topic if does not exist
new_topic = NewTopic(name=get_input_channel(), num_partitions=partition_count, replication_factor=1)
try:
    admin_client.create_topics(new_topics=[new_topic])
except TopicAlreadyExistsError:
    logging.info(f"Topic {get_input_channel()} was already created")

# add Kafka partitions to existing topic
new_partitions = NewPartitions(total_count=partition_count)
try:
    admin_client.create_partitions(topic_partitions={get_input_channel(): new_partitions})
except InvalidPartitionsError as exp:
    logging.info(f"No need to increase Kafka partitions for topic {get_input_channel()}")
</code></pre>
