Kafka Stream 未填充

Question

我有一个非常简单的 Java/Spring 应用程序来演示 KStream 功能，但不幸的是我无法让 KStream 加载数据。想法是创建一个 KStream 对象并使用控制器 GET 方法简单地检索其内容。示例代码：

@RestController
@RequestMapping("/resources/")
public class StreamController {

   private KafkaStreams streams;
   private KStream<String, ResourceMessage> resourceStream;

   StreamController() {

      // configure streams/consumer
      Properties props = new Properties();

      // make sure stream starts from the beginning
      props.put(StreamsConfig.APPLICATION_ID_CONFIG, UUID.randomUUID().toString());
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

      props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
      props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
      props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
      props.put(StreamsConfig.STATE_DIR_CONFIG, Path.of(System.getProperty("java.io.tmpdir")).toAbsolutePath().toString());

      //create POJO serdes
      StreamsBuilder builder = new StreamsBuilder();
      Map<String, Object> serdeProps = new HashMap<>();
      Serializer<ResourceMessage> resourceSerializer = new JsonPOJOSerializer<>();
      serdeProps.put("JsonPOJOClass", ResourceMessage.class);
      resourceSerializer.configure(serdeProps, false);
      Deserializer<ResourceMessage> resourceDeserializer = new JsonPOJODeserializer<>();
      serdeProps.put("JsonPOJOClass", Resource.class);
      resourceDeserializer.configure(serdeProps, false);
      Serde<ResourceMessage> resourceSerde = Serdes.serdeFrom(resourceSerializer, resourceDeserializer);

      // create KStream with POJO serdes for value
      resourceStream = builder.stream("Resources", Consumed.with(Serdes.String(), resourceSerde));
      streams = new KafkaStreams(builder.build(), props);
      streams.start();
     }

   // GET method that enumerates KStream and returns contents
   @GetMapping(value = "/resource")
   public List<Resource> getResources() {

       List<ResourceMessage> messages = new LinkedList<ResourceMessage>();

       // problem is here - there are messages in the topic but KStream returns no values in foreach(...)
       resourceStream.foreach((key, value) -> messages.add(value));

       return messages.stream().map(m -> Resource.builder()
            .id(m.getResouceId()).resource(m.getResource()).build()).collect(Collectors.toList());
   }
}

问题 - 主题中有消息，但 foreach(...) 中的 KStream 枚举没有从中检索到任何结果。 KStream 对象状态为“RUNNING”，日志中没有错误。

生成随机 APPLICATION_ID 并将 AUTO_OFFSET_RESET 设置为“最早”没有帮助。使用 Kafka Tool 可以清楚地看到主题中的一些消息。在控制器运行时添加新消息也无济于事。关于 Kafka 流，我有什么遗漏或不了解的地方吗？

PS 我正在使用来自here 的 POJO 序列化器和反序列化器示例。

Answer 1

Kafka Streams 是一个用于实时流处理的 Kafka 客户端。在您的情况下，您不需要 Kafka Streams 客户端（它不起作用），您需要一个简单的 Kafka Consumer，它从 Kafka 轮询记录并使用 Rest API 将其发回。例如：

@RestController
@RequestMapping("/resources/")
public class StreamController {

   private KafkaStreams streams;
   private Consumer<String, ResourceMessage> consumer;

   StreamController() {

  // configure consumer properties
  Properties props = new Properties();

  // make the right properties with your Serialize and deserialiser

  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

  // Create the consumer using props.
  consumer = new KafkaConsumer<>(props);
  consumer.subscribe(Collections.singletonList(TOPIC));
 }

   // GET method that enumerates KStream and returns contents
   @GetMapping(value = "/resource")
   public List<Resource> getResources() {

   List<ResourceMessage> messages = new LinkedList<ResourceMessage>();

   ConsumerRecords<String, ResourceMessage> consumerRecords =
                consumer.poll(1000);
   
  
   messages = consumerRecords... // Convert the records to your custom POJO

   return messages.stream().map(m -> Resource.builder()
        .id(m.getResouceId()).resource(m.getResource()).build()).collect(Collectors.toList());
   }
}

您可以在此处找到完整示例 link。

更新

此外，您应该知道 RestControlled 是请求范围的，因此会为每个请求创建一个控制器实例。因此，最终您将无法获得 API 响应。如果你想使用 Kafka Streams，你可以在你的 main 方法中启动它，同时你有一个 Spring boot 应用程序。 例如，您可以看到 link。