Styjun

i'm setting up a ConcurrentMessageListenerContainer

<bean class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer" id="messageListenerContainer">
 <constructor-arg index="0" ref="consumerFactory"/>
 <constructor-arg index="1" ref="containerProperties"/>
 <property name="concurrency" value="2"/>
 </bean>

ConsumerFactory use this config:

 <util:map id="consumerConfig" map-class="java.util.HashMap">
 <entry key="#T(org.apache.kafka.clients.consumer.ConsumerConfig).BOOTSTRAP_SERVERS_CONFIG"
 value="$rp.kafka.bootstrap.servers"/>
 <entry key="#T(org.apache.kafka.clients.consumer.ConsumerConfig).KEY_DESERIALIZER_CLASS_CONFIG"
 value="org.apache.kafka.common.serialization.StringDeserializer"/>
 <entry key="#T(org.apache.kafka.clients.consumer.ConsumerConfig).VALUE_DESERIALIZER_CLASS_CONFIG"
 value="org.springframework.kafka.support.serializer.JsonDeserializer"/>
 <entry key="#T(org.springframework.kafka.support.serializer.JsonDeserializer).TRUSTED_PACKAGES"
 value="*"/>
 <entry key="#T(org.apache.kafka.clients.consumer.ConsumerConfig).PARTITION_ASSIGNMENT_STRATEGY_CONFIG"
 value="org.apache.kafka.clients.consumer.RoundRobinAssignor"/>
 <entry key="#T(org.apache.kafka.clients.consumer.ConsumerConfig).ENABLE_AUTO_COMMIT_CONFIG"
 value="false"/>
 </util:map>

and ContainerProperties are

 <bean class="org.springframework.kafka.listener.ContainerProperties" id="containerProperties">
 <constructor-arg>
 <list>
 <value>sendSMS</value>
 </list>
 </constructor-arg>
 <property name="groupId" value="main"/>
 <property name="messageListener" ref="messageListener"/>
 <property name="ackMode" value="RECORD"/>
 </bean>

My topic "sendSMS" has 5 partitions on 3-noded cluster with rep factor of 3, so i expect that each KafkaMessageListenerContainer created by Concurrent one (total 2 in that case) will take it's portion of partitions to handle. Hovewer, after an application is started i see in my debugger window that each listener is handling all 5! partitions
https://gyazo.com/183626ff60061b471858f8cc52573353
and message from 4-th partition (its where i have a message that hangs the processing and not being commited after restarts, but its not related to this issue) on the same offset is being delivered 2 times in different threads with different consumers! Why it happens so? Is it a bug or expected behavior?

You are not showing enough information. The concurrent container aggregates the assigned partitions for the child KafkaListenerContainers (one for each concurrency).

@Override
public Collection<TopicPartition> getAssignedPartitions() 
 return this.containers.stream()
 .map(KafkaMessageListenerContainer::getAssignedPartitions)
 .filter(Objects::nonNull)
 .flatMap(Collection::stream)
 .collect(Collectors.toList());

You need to show logs for the re-delivery; turn on DEBUG logging for more information.