Spring-integration-kafka consumer takes a long time to receive message

I am using spring-integration-kafka in my project, I can create and consume messages normally.

I have 1 producer, 1 high-level customer, 1 section for my theme.

When I sent 5msgs / sec, the consumer takes 1 second to start receiving messages. But when I tried to use multiple threads to send about 300msgs / sec to kafka, I see that it takes a long time for the consumer to start receiving messages, about 5-10 seconds. And larger messages are sent, the slower consumer starts receiving messages. I've also tried adding a few more configurations to reduce the latency, but doesn't seem to have an effect.

So, any idea or suggestion to help me solve this problem? I want to always receive 0-1 sec to start receiving messages. Many thanks.

This is the consumer config:

 <int:channel id="inputFromKafka">
            <int:dispatcher task-executor="kafkaMessageExecutor" />
        </int:channel>

        <int-kafka:zookeeper-connect id="zookeeperConnect"
            zk-connect="127.0.0.1:2181" zk-connection-timeout="6000"
            zk-session-timeout="6000" zk-sync-time="2000" />

        <int-kafka:inbound-channel-adapter
            id="kafkaInboundChannelAdapter" kafka-consumer-context-ref="consumerContext"
            auto-startup="true" channel="inputFromKafka">
            <int:poller fixed-delay="10" time-unit="MILLISECONDS"
                max-messages-per-poll="5" />
        </int-kafka:inbound-channel-adapter>


        <bean id="consumerProperties"
            class="org.springframework.beans.factory.config.PropertiesFactoryBean">
            <property name="properties">
                <props>
                    <prop key="auto.offset.reset">smallest</prop>
                    <prop key="socket.receive.buffer.bytes">1048576</prop>
                    <prop key="fetch.message.max.bytes">5242880</prop>
                    <prop key="auto.commit.interval.ms">10</prop>
                </props>
            </property>
        </bean>

        <int-kafka:consumer-context id="consumerContext"
            consumer-timeout="1000" zookeeper-connect="zookeeperConnect"
            consumer-properties="consumerProperties">
            <int-kafka:consumer-configurations>
                <int-kafka:consumer-configuration
                    group-id="defaultGp" max-messages="10000">
                    <int-kafka:topic id="testTopic" streams="4" />
                </int-kafka:consumer-configuration>
            </int-kafka:consumer-configurations>
        </int-kafka:consumer-context>

        <task:executor id="kafkaMessageExecutor" pool-size="0-10"
            keep-alive="120" queue-capacity="500" />

        <int:outbound-channel-adapter channel="inputFromKafka"
            ref="kafkaConsumer" method="processMessage" />

      

        
        
        
      

    

And the producer:

<int:publish-subscribe-channel id="inputToKafka" />

<int-kafka:outbound-channel-adapter
    id="kafkaOutboundChannelAdapter" kafka-producer-context-ref="kafkaProducerContext"
    auto-startup="true" order="1" channel="inputToKafka" />
<int-kafka:producer-context id="kafkaProducerContext"
    producer-properties="producerProps">
    <int-kafka:producer-configurations>
        <int-kafka:producer-configuration 
            broker-list="127.0.0.1:9092"
            async="true" topic="testTopic"
            key-class-type="java.lang.String" 
            key-encoder="encoder"
            value-class-type="java.lang.String" 
            value-encoder="encoder"
            compression-codec="default" />
    </int-kafka:producer-configurations>
</int-kafka:producer-context>

<util:properties id="producerProps">
    <prop key="queue.buffering.max.ms">500</prop>
    <prop key="topic.metadata.refresh.interval.ms">360000</prop>
    <prop key="queue.buffering.max.messages">10000</prop>
    <prop key="retry.backoff.ms">100</prop>
    <prop key="message.send.max.retries">5</prop>
    <prop key="send.buffer.bytes">5242880</prop>
    <prop key="socket.request.max.bytes">104857600</prop>
    <prop key="socket.receive.buffer.bytes">1048576</prop>
    <prop key="socket.send.buffer.bytes">1048576</prop>
    <prop key="request.required.acks">1</prop>
</util:properties>

<bean id="encoder"
    class="org.springframework.integration.kafka.serializer.common.StringEncoder" />


<task:executor id="taskExecutor" pool-size="5"
    keep-alive="120" queue-capacity="500" />