spring-integration-kafka konfiguracja klienta, aby otrzymać wiadomość od określenia partycji

Question

Zacząłem używać w moim projekcie spring-integration-kafka i mogę produkować i konsumować wiadomości od Kafki. Ale teraz chcę utworzyć komunikat do określonej partycji, a także pobrać wiadomość z określonej partycji.

Przykład chcę produkować wiadomość do strefy 3, a zużywają tylko otrzyma wiadomość z partycji 3.

Do tej pory mój temat ma 8 partycji i mogę produkować wiadomości do określonej partycji, ale mam nie znaleziono sposobu na skonfigurowanie konsumenta, ale otrzymasz wiadomość tylko z konkretnej partycji.

Tak więc wszelkie sugestie dotyczące tego, jak powinienem skonfigurować konsumenta za pomocą wiosennej integracji kafka, lub cokolwiek innego, co trzeba zrobić z klasą KafkaConsumer.java, może odebrać wiadomość z określonej partycji.

Dzięki.

Oto mój kod:

Kafka-producent-context.xml

<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,127.0.0.1:9093,127.0.0.1:9094" 
      async="true" topic="testTopic" 
      key-class-type="java.lang.String" 
      key-encoder="encoder" 
      value-class-type="java.lang.String" 
      value-encoder="encoder" 
      partitioner="partitioner" 
      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">3600000</prop> 
    <prop key="queue.buffering.max.messages">10000</prop> 
    <prop key="retry.backoff.ms">100</prop> 
    <prop key="message.send.max.retries">2</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" /> 

<bean id="partitioner" class="org.springframework.integration.kafka.support.DefaultPartitioner"/> 

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

KafkaProducer.java

public class KafkaProducer { 

private static final Logger logger = LoggerFactory 
     .getLogger(KafkaProducer.class); 

@Autowired 
private MessageChannel inputToKafka; 

public void sendMessage(String message) { 

    try { 
     inputToKafka.send(MessageBuilder.withPayload(message) 
        .setHeader(KafkaHeaders.TOPIC, "testTopic") 
        .setHeader(KafkaHeaders.PARTITION_ID, 3).build()); 
    } catch (Exception e) { 
     logger.error(String.format(
       "Failed to send [ %s ] to topic %s ", message, topic), 
       e); 
    } 
} 

}

Kafka-konsument-context .xml

<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">1000</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="defaultGrp" max-messages="20000"> 
      <int-kafka:topic id="testTopic" streams="3" /> 
     </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" /> 

KafkaConsumer.java

public class KafkaConsumer { 

private static final Logger log = LoggerFactory 
     .getLogger(KafkaConsumer.class); 

@Autowired 
KafkaService kafkaService; 

public void processMessage(Map<String, Map<Integer, List<byte[]>>> msgs) { 
    for (Map.Entry<String, Map<Integer, List<byte[]>>> entry : msgs 
      .entrySet()) { 
     log.debug("Topic:" + entry.getKey()); 
     ConcurrentHashMap<Integer, List<byte[]>> messages = (ConcurrentHashMap<Integer, List<byte[]>>) entry 
       .getValue(); 
     log.debug("\n**** Partition: \n"); 
     Set<Integer> keys = messages.keySet(); 
     for (Integer i : keys) 
      log.debug("p:"+i); 
     log.debug("\n**************\n"); 
     Collection<List<byte[]>> values = messages.values(); 
     for (Iterator<List<byte[]>> iterator = values.iterator(); iterator 
       .hasNext();) { 
      List<byte[]> list = iterator.next(); 
      for (byte[] object : list) { 
       String message = new String(object); 
       log.debug("Message: " + message); 
       try { 
        kafkaService.receiveMessage(message); 
       } catch (Exception e) { 
        log.error(String.format("Failed to process message %s", 
          message)); 
       } 
      } 
     } 

    } 
} 
} 

Więc mój problem jest tutaj. Kiedy generuję komunikat do partycji 3 lub dowolnej partycji, KafkaConsumer zawsze otrzymuje wiadomość. Wszystko, czego chcę, to: KafkaConsumer odbierze tylko wiadomość z partycji 3, a nie z innej partycji.

Jeszcze raz dziękuję.

Answer 1

Musisz użyć message-driven-channel-adapter.

W wariancie KafkaMessageListenerContainer może zaakceptować org.springframework.integration.kafka.core.Partition tablicy argumentu, aby określić tematy i ich partycji parę.

Musisz podłączyć się pojemnik słuchacza, korzystając this constructor i dostarczyć go do zasilacza za pomocą atrybutu listener-container.

Zaktualizujemy plik readme na przykładzie.