使用Jackson生成Avro Schema后注册Confluent Schema Registry失败

问题背景 #

• 已实现Jackson从POJO生成Avro Schema并完成序列化，功能正常
• 测试环境基于Testcontainer搭建，客户端版本7.3.3，Testcontainer镜像为confluentinc/cp-schema-registry:7.3.3
• 更换官方示例Schema、KafkaJsonSchemaSerializer后仍出现相同连接类错误，推测为Schema Registry通用连接问题

核心代码 #

SchemaRegistryClient初始化 #

SchemaRegistryClient schemaRegistryClient( final String url ) {
    return new CachedSchemaRegistryClient( url, 100 );
}

序列化及Schema上传逻辑 #

public byte[] serialize( final Optional<Object> root ) {
    if ( root.isEmpty() ) {
        return new byte[0];
    }
    final Object rootObject = root.get();
    final String subjectName = rootObject.getClass().getSimpleName();

    try {
        // Jackson Avro Schema
        final AvroSchema schema = getAvroSchema( rootObject );
        final io.confluent.kafka.schemaregistry.avro.AvroSchema avroSchema = new io.confluent.kafka.schemaregistry.avro.AvroSchema( schema.getAvroSchema(),
              1 );
        schemaRegistryClient.register( subjectName, avroSchema, false );
        return avroMapper.writer( schema ).writeValueAsBytes( rootObject );
    } catch ( final IOException | RuntimeException e ) {
        throw new SerializationException( "Error serializing Avro message", e );
    } catch ( final RestClientException e ) {
        throw toKafkaException( e );
    }
}

错误信息 #

第一种URL获取方式 #

public String getClientUrl() {
    return String.format( "http://%s:%s", getHost(), getMappedPort( SCHEMA_REGISTRY_PORT ) );
}

错误日志：

java.net.SocketException: Unexpected end of file from server
    at java.base/sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:903)
    at java.base/sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:726)
    at java.base/sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:900)
    at java.base/sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:726)
    at java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream0(HttpURLConnection.java:1688)
    at java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1589)
    at java.base/java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:529)
    at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:294)
    at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:384)
    at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:561)
    at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:549)
    at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.registerAndGetId(CachedSchemaRegistryClient.java:290)
    at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:397)
    at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:370)

第二种URL获取方式 #

public String getClientUrl() {
    return String.format( "http://%s:%s", getHost(), getExposedPorts().get( 0 ) );
}

错误日志：

java.net.ConnectException: Connection refused

生成的Avro Schema示例 #

{
  "type": "record",
  "name": "TestDto",
  "namespace": "test.model",
  "fields": [
    {
      "name": "id",
      "type": [
        "null",
        {
          "type": "long",
          "java-class": "java.lang.Long"
        }
      ]
    },
    {
      "name": "labels",
      "type": [
        "null",
        {
          "type": "array",
          "items": "string",
          "java-class": "java.util.Set"
        }
      ]
    }
  ]
}

解决方案建议 #

• 校验Schema Registry容器配置：
  • 确保容器启动时设置SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS指向Testcontainer中Kafka的正确地址
  • 确认SCHEMA_REGISTRY_LISTENERS配置为http://0.0.0.0:8081，允许外部访问
• 修正URL获取逻辑：
  • 保留getMappedPort(SCHEMA_REGISTRY_PORT)的方式，确保SCHEMA_REGISTRY_PORT常量值为容器内部默认端口8081
  • 避免使用getExposedPorts().get(0)，容器可能暴露多个端口，无法保证第一个是Schema Registry端口
• 添加容器启动等待逻辑：
  • Schema Registry启动后需与Kafka建立连接完成初始化，可添加健康检查，调用其/subjects端点确认服务可用后再执行上传
• 排查依赖兼容性：
  • 检查项目中io.confluent相关依赖版本是否与容器版本7.3.3完全一致，排除依赖冲突
• 验证网络连通性：
  • 在测试代码中添加HTTP请求测试，确认能正常访问http://<host>:<mapped-port>/subjects端点，排除网络层面问题

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