Spring Cloud Stream集成Confluent Schema Registry异常求助

Let's walk through the key issues in your setup and fix them step by step:

1. Critical Hardcoding Bug in Consumer Configuration #

First, your consumer's schema registry client bean is ignoring the configured endpoint and hardcoding http://localhost:8081 directly. This means even if you change the endpoint in your config file, the consumer will never pick it up. Fix this immediately:

@Configuration
static class ConfluentSchemaRegistryConfiguration {
    @Bean
    public SchemaRegistryClient schemaRegistryClient(@Value("${spring.cloud.stream.schemaRegistryClient.endpoint}") String endpoint){
        ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient();
        client.setEndpoint(endpoint); // Use the injected config value, not hardcoded
        // Add timeouts to trigger errors if connection fails
        client.setConnectTimeout(5000);
        client.setReadTimeout(10000);
        return client;
    }
}

2. Verify Dependency Setup #

Make sure you have the correct dependencies in your build file (Maven/Gradle) to enable Schema Registry integration:
For Maven:

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-schema-registry-client</artifactId>
</dependency>
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-schema-registry-client</artifactId>
    <version>${confluent.version}</version> <!-- Match your Confluent Platform version -->
</dependency>

Missing or mismatched dependencies can silently break Schema Registry communication without throwing obvious errors.

3. Validate Schema Registry Reachability #

Before blaming your app, confirm the Schema Registry is actually accessible from your application's environment:

• Run this command from the same machine/container where your app runs:

  curl http://localhost:8081/subjects
  

• If you get a JSON response (even an empty array), the registry is up. If you get a connection error:
  • Check if the registry is running on the correct port
  • If using Docker, ensure you're not using localhost (use host.docker.internal for Docker Desktop, or your host machine's IP)
  • Verify no firewalls/security groups are blocking the port

4. Enable Debug Logging to Uncover Silent Failures #

Your app isn't throwing timeout errors because the issue might be getting swallowed. Add these log settings to your application.yml to see detailed Schema Registry communication:

logging:
  level:
    org.springframework.cloud.stream.schema: DEBUG
    io.confluent.kafka.schemaregistry: DEBUG
    org.springframework.cloud.stream.binder.kafka: DEBUG

Look for logs about schema registration/fetch attempts—this will show if requests are being sent, and if there are hidden errors like authentication issues or schema validation failures.

5. Clean Up Configuration Details #

• Fix Producer Content-Type: Use a specific Avro content type instead of the wildcard:

  contentType: application/vnd.apache.avro+json
  

• Validate YAML Indentation: Ensure your config file has correct indentation (YAML is strict about this). Your current config looks right, but double-check that schemaRegistryClient is nested under spring.cloud.stream.

6. Confirm Topic and Schema Alignment #

• Ensure the sensor-topic exists in Kafka (create it manually if needed)
• Verify that the sensor.avsc schema file is correctly placed in classpath:avro/ and matches the generated Sensor class structure.

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