在 Kafka Consumer API 中实现 Deserializer 和 Serde 有什么区别？

Question

我尝试在 GitHub (https://github.com/onurtokat/kafka-clickstream-enrich) 上模拟 Gwen (Chen) Shapira 的 kafka-clickstream-enrich kafka-stream 项目。当我使用反序列化器使用 consumer class 消费主题时，遇到错误。定制的 Serde class 有序列化器和反序列化器。但是，我试图理解为什么自定义 serde 用于反序列化器，然后消费者 API 给出错误，因为它不是 org.apache.kafka.common.serialization.Deserializer

的实例

可以使用带有 Serdes.Integer() 序列化程序和新的 ProfileSerde() 反序列化程序的 KTable 使用该主题，如下所示。

KTable<Integer, UserProfile> profiles = builder.table(Constants.USER_PROFILE_TOPIC,
                Consumed.with(Serdes.Integer(), new ProfileSerde()),
                Materialized.as("profile-store"));

自定义 Serde 定义为；

static public final class ProfileSerde extends WrapperSerde<UserProfile> {
        public ProfileSerde() {
            super(new JsonSerializer<UserProfile>(), new JsonDeserializer<UserProfile>(UserProfile.class));
        }
    }

通用 Serde 是自定义的，如下所示；

package com.onurtokat.serde;

import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;

import java.util.Map;


public class WrapperSerde<T> implements Serde<T> {

    final private Serializer<T> serializer;
    final private Deserializer<T> deserializer;

    public WrapperSerde(Serializer<T> serializer, Deserializer<T> deserializer) {
        this.serializer = serializer;
        this.deserializer = deserializer;
    }

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        serializer.configure(configs, isKey);
        deserializer.configure(configs, isKey);
    }

    @Override
    public void close() {
        serializer.close();
        deserializer.close();
    }

    @Override
    public Serializer<T> serializer() {
        return serializer;
    }

    @Override
    public Deserializer<T> deserializer() {
        return deserializer;
    }
}

我的Consumer就这么简单，可以看下面；

package com.onurtokat.consumers;

import com.onurtokat.ClickstreamEnrichment;
import com.onurtokat.Constants;
import com.onurtokat.model.UserProfile;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class ConsumeProfileData {

    public static void main(String[] args) {
        //prepare config
        Properties config = new Properties();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ClickstreamEnrichment.ProfileSerde.class);

        KafkaConsumer<Integer, UserProfile> consumerProfileTopic = new KafkaConsumer<>(config);
        consumerProfileTopic.subscribe(Arrays.asList(Constants.USER_PROFILE_TOPIC));
        while (true) {
            ConsumerRecords<Integer, UserProfile> records = consumerProfileTopic.poll(Duration.ofMillis(100));
            for (ConsumerRecord<Integer, UserProfile> record : records) {
                System.out.println(record.key() + " " + record.value());
            }
        }
    }
}

当我尝试与我的消费者一起使用主题时的错误是；

log4j:WARN No appenders could be found for logger (org.apache.kafka.clients.consumer.ConsumerConfig).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Exception in thread "main" org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:811)
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:659)
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:639)
    at com.onurtokat.consumers.ConsumeProfileData.main(ConsumeProfileData.java:25)
Caused by: org.apache.kafka.common.KafkaException: com.onurtokat.ClickstreamEnrichment$ProfileSerde is not an instance of org.apache.kafka.common.serialization.Deserializer
    at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:304)
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:712)
    ... 3 more

Answer 1

你好像误会了：

The topic can be consumed using KTable with Serdes.Integer() Serializer and new ProfileSerde() Deserializer like below.

您必须 Consumed.with() 提供 KeySerde 和 ValueSerde。

关于异常：

很明显 - 您必须设置 Deserializer 的实现（不是 Serde)

config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, //here);

Answer 2

区别是：

Serdes 被 Kafka 的 Streams API（又名 Kafka Streams）使用。 Serde 是一对相同数据类型的 (1) serializer 和 (2) deserializer 的包装器——请参阅接下来的两个要点。也就是说，一个 Serde<T> 有一个 Serializer<T> 和一个 Deserializer<T>。您发布的第一个代码片段（例如 KTable）是一个 Kafka Streams 代码片段，这就是它需要 Serde 的原因。 Kafka Streams 需要 Serde，因为它既会生成消息（为此需要 Serializer）又会读取消息（为此需要 Deserializer）。
反序列化器 被 Kafka 的消费者 API（又名消费者客户端）用于阅读消息。您的最后一个代码片段（例如 KafkaConsumer）正在使用消费者客户端，因此需要 Deserializer，而不是 Serde。
序列化程序 被 Kafka 的生产者 API（又名生产者客户端）用于写入消息。

关于：

Caused by: org.apache.kafka.common.KafkaException: com.onurtokat.ClickstreamEnrichment$ProfileSerde is not an instance of org.apache.kafka.common.serialization.Deserializer
    at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:304)
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:712)
    ... 3 more

您的 Kafka 消费者客户端代码被赋予 Serde，而它期望 Deserializer。

在 Kafka Consumer API 中实现 Deserializer 和 Serde 有什么区别？

What is the difference between implementing Deserializer and Serde in Kafka Consumer API?

java

apache-kafka

kafka-consumer-api

apache-kafka-streams