Kafka Streams无法连接Upstash Schema Registry排查求助

问题背景 #

我的Kafka Streams应用无法连接Upstash Schema Registry，但通过curl请求目标Registry地址能正常获取Schema数据，说明问题出在应用配置或代码中。

curl验证结果 #

执行curl请求：

curl https://right-boa-11231-eu1-rest-kafka.upstash.io/schema-registry/schemas/ids/8?fetchMaxId=false&subject=test1-value

返回结果：

{"schema": "{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"fr.potato\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"favorite_number\",\"type\":\"long\"},{\"name\":\"timestamp_string\",\"type\":\"string\"}]}"
}

应用抛出的异常 #

2024-07-15 12:50:00 DEBUG StreamThread:1201 - stream-thread [enrichement-app-14-7e65669d-662f-44a6-b47a-30af74085b4e-StreamThread-1] Main Consumer poll completed in 133 ms and fetched 1 records from partitions [test1-0]
2024-07-15 12:50:00 DEBUG RestService:292 - Sending GET with input null to https://right-boa-11231-eu1-rest-kafka.upstash.io/schema-registry/schemas/ids/8?fetchMaxId=false&subject=test1-value
2024-07-15 12:50:00 ERROR LogAndFailExceptionHandler:39 - 
Exception
 caught during Deserialization, taskId: 0_0, topic: test1, partition: 0, offset: 0
org.apache.kafka.common.errors.SerializationException
: Error retrieving Avro value schema for id 8
  at io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe.toKafkaException(AbstractKafkaSchemaSerDe.java:805) ~[kafka-schema-serializer-7.6.1.jar:?]
  at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.schemaFromRegistry(AbstractKafkaAvroDeserializer.java:415) ~[kafka-avro-serializer-7.6.1.jar:?]
  at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:188) ~[kafka-avro-serializer-7.6.1.jar:?]
  at io.confluent.kafka.serializers.KafkaAvroDeserializer.deserialize(KafkaAvroDeserializer.java:107) ~[kafka-avro-serializer-7.6.1.jar:?]
  at io.confluent.kafka.serializers.KafkaAvroDeserializer.deserialize(KafkaAvroDeserializer.java:102) ~[kafka-avro-serializer-7.6.1.jar:?]
  at io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer.deserialize(GenericAvroDeserializer.java:63) ~[kafka-streams-avro-serde-5.2.1.jar:?]
  at io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer.deserialize(GenericAvroDeserializer.java:39) ~[kafka-streams-avro-serde-5.2.1.jar:?]
  at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:62) ~[kafka-clients-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:58) ~[kafka-streams-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66) [kafka-streams-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.RecordQueue.updateHead(RecordQueue.java:204) [kafka-streams-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:128) [kafka-streams-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:285) [kafka-streams-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:1039) [kafka-streams-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.TaskManager.addRecordsToTasks(TaskManager.java:1782) [kafka-streams-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.StreamThread.pollPhase(StreamThread.java:1208) [kafka-streams-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.StreamThread.runOnceWithoutProcessingThreads(StreamThread.java:909) [kafka-streams-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:686) [kafka-streams-3.7.1.jar:?]
  at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:645) [kafka-streams-3.7.1.jar:?]
Caused by: 
io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException
: null; error code: 0
  at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:336) ~[kafka-schema-registry-client-7.6.1.jar:?]
  at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:409) ~[kafka-schema-registry-client-7.6.1.jar:?]
  at io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:916) ~[kafka-schema-registry-client-7.6.1.jar:?]
  at io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:900) ~[kafka-schema-registry-client-7.6.1.jar:?]
  at io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:880) ~[kafka-schema-registry-client-7.6.1.jar:?]
  at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaByIdFromRegistry(CachedSchemaRegistryClient.java:333) ~[kafka-schema-registry-client-7.6.1.jar:?]
  at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaBySubjectAndId(CachedSchemaRegistryClient.java:464) ~[kafka-schema-registry-client-7.6.1.jar:?]
  at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.schemaFromRegistry(AbstractKafkaAvroDeserializer.java:398) ~[kafka-avro-serializer-7.6.1.jar:?]
  ... 17 more

应用代码 #

package fr.potato;

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

import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
import io.confluent.kafka.streams.serdes.avro.GenericAvroSerde;
import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class Enrichement {
  static String auth = "username:pw";
  static String sourceTopicName = "test1";
  static String targetTopicName = "test2";
  static String schemaRegistryUrl = "https://right-boa-11231-eu1-rest-kafka.upstash.io/schema-registry";
  private static final Logger logger = LogManager.getLogger(Enrichement.class);

  // @SuppressWarnings("deprecation")
  public static void main(String[] args) {
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "enrichement-app-14");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "https://right-boa-11231-eu1-kafka.upstash.io:9092");
    props.put("sasl.mechanism", "SCRAM-SHA-256");
    props.put("security.protocol", "SASL_SSL");
    props.put("sasl.jaas.config",
        "org.apache.kafka.common.security.scram.ScramLoginModule required=\"username\" password=\"pw\";";");
    props.put("basic.auth.credentials.source", "USER_INFO");
    props.put("basic.auth.user.info", auth);
    // https://right-boa-11231-eu1-rest-kafka.upstash.io/schema-registry/schemas/ids/8?fetchMaxId=false&subject=test1-value
    props.put("schema.registry.url", "right-boa-11231-eu1-rest-kafka.upstash.io/schema-registry");
    props.put("debug", "true");
    props.put("ssl.endpoint.identification.algorithm", "https");

    props.put("key.converter", "org.apache.kafka.connect.storage.StringConverter");
    props.put("value.converter", "io.confluent.connect.avro.AvroConverter");
    props.put("value.converter.schema.registry.url", "right-boa-11231-eu1-rest-kafka.upstash.io/schema-registry");

    props.put("key.converter.basic.auth.credentials.source", "USER_INFO");
    props.put("key.converter.basic.auth.user.info", auth);
    props.put("value.converter.basic.auth.credentials.source", "USER_INFO");
    props.put("value.converter.basic.auth.user.info", auth);

    props.put("auto.register.schemas", false);
    props.put("use.latest.version", true);

    final Map<String, String> serdeConfig = Collections.singletonMap(
        "schema.registry.url", schemaRegistryUrl);

    final Serde<GenericRecord> valueGenericAvroSerde = new GenericAvroSerde();
    valueGenericAvroSerde.configure(serdeConfig, false); // `false` for record values

    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, GenericRecord> inputStream = builder.stream(sourceTopicName,
        Consumed.with(Serdes.String(), valueGenericAvroSerde));

    inputStream
        .peek((key, value) -> System.out.println("Key: " + key + ", Value: " + value))
        .to(targetTopicName, Produced.with(Serdes.String(), valueGenericAvroSerde));

    try {
      KafkaStreams streams = new KafkaStreams(builder.build(), props);
      streams.setUncaughtExceptionHandler((Thread thread, Throwable throwable) -> {
        logger.error("Uncaught exception in thread " + thread, throwable);
      });
      streams.start();
      System.out.println("Kafka Streams app started successfully.");

      Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    } catch (Exception e) {
      System.err.println("Error starting Kafka Streams app: " + e.getMessage());
      e.printStackTrace();
    }
  }
}

排查原因及解决办法 #

1. 修复Schema Registry URL的协议前缀 #

代码中props.put("schema.registry.url", "right-boa-11231-eu1-rest-kafka.upstash.io/schema-registry");和props.put("value.converter.schema.registry.url", "right-boa-11231-eu1-rest-kafka.upstash.io/schema-registry");缺少https://前缀，导致客户端无法正确发起HTTPS请求。

修改为：

props.put("schema.registry.url", schemaRegistryUrl); // 直接使用已定义的带https的变量
props.put("value.converter.schema.registry.url", schemaRegistryUrl);

2. 为Serde配置添加认证信息 #

Upstash Schema Registry需要Basic Auth认证，但当前serdeConfig仅配置了URL，未包含认证参数，导致Serde发起请求时无权限访问Registry。

修改serdeConfig：

final Map<String, String> serdeConfig = new HashMap<>();
serdeConfig.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
serdeConfig.put(AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
serdeConfig.put(AbstractKafkaSchemaSerDeConfig.USER_INFO_CONFIG, auth);

3. 修正SASL JAAS配置的语法错误 #

当前sasl.jaas.config存在语法错误：末尾多了一个分号和引号，且required后不需要加引号包裹参数。

修改为：

props.put("sasl.jaas.config",
    "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"username\" password=\"pw\";");

4. 统一Confluent依赖版本 #

代码中使用的kafka-streams-avro-serde版本为5.2.1，而kafka-schema-registry-client版本为7.6.1，版本差异过大可能导致兼容性问题。将所有Confluent相关依赖统一为同一版本（建议使用与Kafka版本匹配的Confluent版本，Kafka 3.7.1对应Confluent 7.6.x）。

总结 #

通过修复URL前缀、补充认证配置、修正语法错误及统一依赖版本，即可解决Kafka Streams应用与Upstash Schema Registry的连接问题。

内容的提问来源于stack exchange，提问作者Carl