我的Kafka流媒体应用程序退出，代码为0，什么也不做 [英] My Kafka streaming application just exit with code 0 doing nothing

查看：22
本文介绍了我的Kafka流媒体应用程序退出，代码为0，什么也不做的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！

问题描述

为了尝试Kafka流，我这样做了：

public static void main(String[] args) {

        final StreamsBuilder builder = new StreamsBuilder();

        final Properties streamsConfiguration = new Properties();

        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "generic-avro-integration-test");
        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, Utils.BOOTSTRAP_SERVER);
        streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, StringDeserializer.class);
        streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, KafkaAvroDeserializer.class);
        streamsConfiguration.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, Utils.SCHEMA_REGISTRY_URL);
        
        builder.stream(Utils.ALL_FX_EVENTS_TOPIC).foreach((key, value) -> System.out.println(key));
        
        KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
        kafkaStreams.start();
    }

但当我在本地运行它时，我只得到以下结果：

基本上我从我的IDE运行它，1秒后它就停止了，而它应该正在等待推送到主题中的新事件。

我不明白。

Kafka主题在另一台计算机上，但我也编写了一个非常简单的使用者的代码，并且我能够阅读来自此远程主题的消息。

出于某种原因，这个非常简单的Kafka流应用程序退出，代码为0。 我无能为力，你知道吗？

由于该问题似乎与此处的slf4j依赖项有关，因此pom：

4.0.0 广口瓶 
<name>Ingestor :: Bigdata :: Ingestor</name>
<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>8</source>
                <target>8</target>
            </configuration>
        </plugin>
    </plugins>
</build>

<parent>
    <groupId>com.parent-pom</groupId>
    <artifactId>parent-pom</artifactId>
    <version>4.0.2</version>
</parent>

<groupId>com</groupId>
<artifactId>ingestor</artifactId>
<version>1.0.1-SNAPSHOT</version>


<properties>
    
    <sq.artifact.type>internal</sq.artifact.type>
    <maven-compiler-plugin.version>3.1</maven-compiler-plugin.version>
    <maven-assembly-plugin.version>3.3.0</maven-assembly-plugin.version>
    <maven.compiler.source>7</maven.compiler.source>
    <maven.compiler.target>7</maven.compiler.target>
    <revision>1.0.0-SNAPSHOT</revision>
    <sq.scs>fx-dan</sq.scs>
</properties>

<dependencies>
    <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>3.1.0</version>
    </dependency>
    <dependency>
        <groupId>com</groupId>
        <artifactId>libs-schemas</artifactId>
        <version>1.0.6-SNAPSHOT</version>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-avro-serializer</artifactId>
        <version>7.0.1</version>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-schema-registry-client</artifactId>
        <version>7.0.1</version>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>common-config</artifactId>
        <version>7.0.1</version>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>common-utils</artifactId>
        <version>7.0.1</version>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-schema-serializer</artifactId>
        <version>7.0.1</version>
    </dependency>
    <dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>31.0.1-jre</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
    </dependency>
</dependencies>

更新：

除了缺少log4j属性文件之外，错误还是由于缺少Serdes的配置。

更新后的代码如下所示：

public static void main(String[] args) {

        final StreamsBuilder builder = new StreamsBuilder();

        final Properties streamsConfiguration = new Properties();
        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "generic-avro-integration-test");
        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, Utils.BOOTSTRAP_SERVER);
        streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
        streamsConfiguration.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, Utils.SCHEMA_REGISTRY_URL);

        final Serde<String> stringSerde = Serdes.String();
        final Serde<AllTypesFxEvents> specificAvroSerde = new SpecificAvroSerde<>();

        final boolean isKeySerde = false;
        specificAvroSerde.configure(Collections.singletonMap(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, Utils.SCHEMA_REGISTRY_URL),
                isKeySerde);

builder.stream(Utils.ALL_FX_EVENTS_TOPIC).foreach((key, value) -> System.out.println(key));

KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
        kafkaStreams.cleanUp();
        kafkaStreams.start();

我不再有异常堆栈，但此链接帮助我修复了它。

proper guide for java kafka stream with avro schema registry

推荐答案

我已修复问题，以下是摘要。

第一个问题是我没有正确设置slf4j。Kafka希望slf4j可用，因为它使用slf4j来打印遇到的错误。没有它，我的程序就会默默不及格。

若要解决此问题，您需要在Resources文件夹(在src文件夹中)中添加一个log4j.properties文件。

我的是这样的(它不打印调试日志)：

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
一旦我有了日志，我就有了关于序列化错误的堆栈跟踪。我不再有日志了，但另一篇让我修复它的帖子是：

Use Kafka Streams with Avro Schema Registry

基本上，我对Serde配置使用了错误的对象。

更新后的代码如下所示：

PUBLIC STATIC VID Main(字符串[]args){

    final StreamsBuilder builder = new StreamsBuilder();

    final Properties streamsConfiguration = new Properties();
    streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "generic-avro-integration-test");
    streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, Utils.BOOTSTRAP_SERVER);
    streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
    streamsConfiguration.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, Utils.SCHEMA_REGISTRY_URL);

    final Serde<String> stringSerde = Serdes.String();
    final Serde<AllTypesFxEvents> specificAvroSerde = new SpecificAvroSerde<>();

    final boolean isKeySerde = false;
    specificAvroSerde.configure(Collections.singletonMap(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, Utils.SCHEMA_REGISTRY_URL),
            isKeySerde);

builder.stream(Utils.ALL_FX_EVENTS_TOPIC).foreach((key, value) -> System.out.println(key));

KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
        kafkaStreams.cleanUp();
        kafkaStreams.start();

这篇关于我的Kafka流媒体应用程序退出，代码为0，什么也不做的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！

查看全文
相关文章
Java开发最新文章
热门教程
热门工具
登录 关闭
扫码关注1秒登录
发送“验证码”获取 | 15天全站免登陆