Spring Kafka流处理器不工作

Question

我正试图使用Spring引导编写一个Kafka流处理器，但当消息生成到主题中时，它不会被调用。

我有以下制作人，它与主题名称adt.events.location配合得很好。

@Service
public class Producer {
private static final Logger logger = LoggerFactory.getLogger(Producer.class);
private static final String TOPIC = "adt.events.location";
private final KafkaTemplate<String, Object> kafkaTemplate;
public Producer(KafkaTemplate<String, Object> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
@EventListener(ApplicationStartedEvent.class)
public void produce() {
this.kafkaTemplate.send(TOPIC, "1", new EventPatientCheckedIn(1L, 1001L, 104L, 11L, 6L));
this.kafkaTemplate.send(TOPIC, "1", new EventPatientBedChanged(1L, 1001L, 7L));
this.kafkaTemplate.send(TOPIC, "1", new EventPatientRoomChanged(1L, 1001L, 10L));
this.kafkaTemplate.send(TOPIC, "2", new EventPatientCheckedIn(2L, 1002L, 110L, 18L, 2L));
this.kafkaTemplate.send(TOPIC, "3", new EventPatientCheckedIn(3L, 1003L, 111L, 16L, 1L));
this.kafkaTemplate.send(TOPIC, "1", new EventPatientCheckedOut(1L, 1001L));
this.kafkaTemplate.send(TOPIC, "3", new EventPatientBedChanged(3L, 1003L, 3L));
}
}

主题消息有不同的类型，并且采用Avro格式。架构在架构注册表中使用Avro union注册

以下是的主题

@Configuration
public class KafkaConfig {
@Bean
public NewTopic topicEventsLocation() {
return TopicBuilder.name("adt.events.location").partitions(1).replicas(1).build();
}
@Bean
public NewTopic topicPatientLocation() {
return TopicBuilder.name("adt.patient.location").partitions(1).replicas(1).build();
}
}

application.yml我在一个社区中使用cp-all作为docker文件

server:
port: 9000
spring:
kafka:
properties:
auto:
register:
schemas: false
use:
latest:
version: true
schema:
registry:
url: http://localhost:8081
consumer:
bootstrap-servers: localhost:9092
group-id: group_id
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
producer:
bootstrap-servers: localhost:9092
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
streams:
application-id: kafka-demo
properties:
default.key.serde: org.apache.kafka.common.serialization.Serdes$LongSerde
default.value.serde: io.confluent.kafka.streams.serdes.avro.GenericAvroSerde

Spring Boot应用程序

@SpringBootApplication
@EnableKafkaStreams
public class KafkaDemoApplication {
public static void main(String[] args) {
SpringApplication.run(KafkaDemoApplication.class, args);
}
}

和处理器

@Component
public class Processor {
final StreamsBuilder builder = new StreamsBuilder();
@Autowired
public void process() {
...    
}
}

预期的行为是，每次生成消息时都会打印出模式名称，并使用流处理输出填充输出主题adt.patient.location，但什么也没发生。

我是卡夫卡的新手，所以我可能错过了什么。

更新

实际上我错过了@EnableKafkaStreams注释。

但现在我得到以下错误：

2021-04-07 16:02:16.967 ERROR 120225 --- [           main] org.apache.kafka.streams.KafkaStreams    : stream-client [LocationService-9611eedf-df9b-4fe5-9a7d-058027cee22a] Topology with no input topics will create no stream threads and no global thread.
2021-04-07 16:02:16.967  WARN 120225 --- [           main] ConfigServletWebServerApplicationContext : Exception encountered during context initialization - cancelling refresh attempt: org.springframework.context.ApplicationContextException: Failed to start bean 'defaultKafkaStreamsBuilder'; nested exception is org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is org.apache.kafka.streams.errors.TopologyException: Invalid topology: Topology has no stream threads and no global threads, must subscribe to at least one source topic or global table.

Answer 1

在KafkaTemplate上使用@Autowired。我想这就是你所缺少的东西。我给出的示例没有使用AvroSerializer。所以我假设您的序列化程序正在工作。至少您应该看到到达使用者的消息或序列化错误。此外，您可以改进处理回调的方法，并使用更一致的消息记录。例如，使用ProducerRecord创建要发送的消息。使用ListenableFuture添加回调。

@Slf4j
@Service
public class Producer {
@Autowired
KafkaTemplate<String, Object> kafkaTemplate;
public void produce() {
String key = "1";
Object value = EventPatientCheckedIn....
ProducerRecord<String, Object> producerRecord = buildProducerRecord(TOPIC, key, value);
ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaTemplate.send(producerRecord);
listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
@Override
public void onFailure(Throwable ex) { handleFailure(key, value, ex); }
@Override
public void onSuccess(SendResult<String, Object> result) { handleSuccess(key, value, result); }
});
}
private ProducerRecord<String, Object> buildProducerRecord(String topic, String key, Object value) {
List<Header> recordHeaders = List.of(new RecordHeader("event-source", "scanner".getBytes()));
return new ProducerRecord<String, Object>(topic, null, key, value, recordHeaders);
}
private void handleSuccess(String key, Object value, SendResult<String, Object> result) {
log.info("message send successfully for the key: {} and value: {} at partition: {}", key, value, result.getRecordMetadata().partition());
}
private void handleFailure(String key, Object value, Throwable ex) {
log.error("error sending the message and the exception us {}", ex.getMessage());
try { throw ex; }
catch (Throwable throwable) {
log.error("error on failure: {}", throwable.getMessage());
}
}
}

更新：我认为您缺少配置Properties，然后在Processor上制作streams.start();。我把这个例子建立在这个参考文献的基础上。

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "adt.events.location");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, YOUR_AVRO_SERDE_HERE);
...
...
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Answer 2

我发现两个错误：

1. 添加注释@EnableKafkaStreams
2. StreamsBuilder注入处理器出错

@SpringBootApplication
@EnableKafkaStreams
public class KafkaDemoApplication {
public static void main(String[] args) {
SpringApplication.run(KafkaDemoApplication.class, args);
}
}

@Autowired
public void process(final StreamsBuilder builder) {
logger.info("Processing location events");
...
}