更新（5.5及更新版本）
汇合型 5.5 通过添加对基本avro类型的本机支持 PrimitiveAvroSerde （参见。https://github.com/confluentinc/schema-registry/blob/5.5.x/avro-serde/src/main/java/io/confluent/kafka/streams/serdes/avro/primitiveavroserde.java)
原始答案（版本5.4及更高版本）：
这是一个众所周知的问题。原始的avro类型不能很好地处理confluent的avroserdes，因为serdes可以处理 GenericAvroRecord 以及 SpecificAvroRecord 只是。
比较https://github.com/confluentinc/schema-registry/tree/master/avro-serde/src/main/java/io/confluent/kafka/streams/serdes/avro.
因此，基于 KafkaAvroSerializer 以及 KafkaAvroDeserializer 是正确的方法。为了能够将其作为默认serde传递到配置中，您不能使用 Serdes.serdeFrom 因为类型信息由于genrics类型擦除而丢失。
但是，您可以实现自己的扩展类 Serde 接口，并将自定义类传递到配置中：

public class MySerde extends Serde<Integer> {
    // use KafkaAvroSerializer and KafkaAvroDeserializer and cast `Object` to `Integer`
}

config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, MySerde.class);

感谢@matthias j.的提示。sax，我想发布解决方案的工作。请免费增强它。

import java.util.Collections;
import java.util.Map;

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

import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;

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

    private final Serde<Object> inner;

    /**
     * Constructor used by Kafka Streams.
     */
    public GenericPrimitiveAvroSerDe() {
        inner = Serdes.serdeFrom(new KafkaAvroSerializer(), new KafkaAvroDeserializer());
    }

    public GenericPrimitiveAvroSerDe(SchemaRegistryClient client) {
        this(client, Collections.emptyMap());
    }

    public GenericPrimitiveAvroSerDe(SchemaRegistryClient client, Map<String, ?> props) {
        inner = Serdes.serdeFrom(new KafkaAvroSerializer(client), new KafkaAvroDeserializer(client, props));
    }

    @Override
    public void configure(final Map<String, ?> serdeConfig, final boolean isSerdeForRecordKeys) {
        inner.serializer().configure(serdeConfig, isSerdeForRecordKeys);
        inner.deserializer().configure(serdeConfig, isSerdeForRecordKeys);
    }

    @Override
    public void close() {
        // TODO Auto-generated method stub
        inner.serializer().close();
        inner.deserializer().close();

    }

    @SuppressWarnings("unchecked")
    @Override
    public Serializer<T> serializer() {
        // TODO Auto-generated method stub
        Object obj = inner.serializer();
        return (Serializer<T>) obj;

    }

    @SuppressWarnings("unchecked")
    @Override
    public Deserializer<T> deserializer() {
        // TODO Auto-generated method stub
        Object obj = inner.deserializer();
        return (Deserializer<T>) obj;

    }

}

用作默认流配置：

props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,GenericPrimitiveAvroSerDe.class);
            props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,GenericPrimitiveAvroSerDe.class);

覆盖默认值：

final Map<String, String> serdeConfig = Collections.singletonMap("schema.registry.url",
                                                                        "http://localhost:8081");
       final GenericPrimitiveAvroSerDe<String> keyGenericAvroSerde = new GenericPrimitiveAvroSerDe<String>();
       keyGenericAvroSerde.configure(serdeConfig, true); // `true` for record keys
       final GenericPrimitiveAvroSerDe<Long> valueGenericAvroSerde = new GenericPrimitiveAvroSerDe<Long>();
       valueGenericAvroSerde.configure(serdeConfig, false); // `false` for record values

如果你想和Kafka的消费者和制作人一起使用@thiyaga rajan这个伟大的解决方案

consumerConfig.put(KEY_DESERIALIZER_CLASS_CONFIG,GenericPrimitiveAvroSerDe.class);
producerConfig.put(KEY_SERIALIZER_CLASS_CONFIG, GenericPrimitiveAvroSerDe.class);

将此添加到类中

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

    public class GenericPrimitiveAvroSerDe<T> implements Serde<T>, Serializer<T>, Deserializer<T> {

      @Override
      public T deserialize(String topic, byte[] data) {
        return this.deserializer().deserialize(topic, data);
      }

      @Override
      public byte[] serialize(String topic, T data) {
        return this.serializer().serialize(topic, data);
      }

    ...

    }