Confluent 4.0.0 Kafka Connect Schema Registry主题未找到，任务失败求助

Let's break down your issue and walk through actionable steps to fix the org.apache.kafka.connect.errors.DataException: emailfilters error (rooted in "Schema Registry Subject Not Found") you're seeing with your Elasticsearch Sink connector.

First, Understand the Root Cause #

Your sink task is failing because the Avro Converter can't find the schema for the emailfilters topic in your Schema Registry. By default, Kafka Connect's Avro Converter looks for subjects named <topic-name>-value (for message values) or <topic-name>-key (for message keys) in the Schema Registry. The error points to emailfilters, so it's likely searching for either emailfilters-value or emailfilters-key and coming up empty.

Step-by-Step Fixes #

1. Verify the Subject Exists in Schema Registry

First, check if the expected subject is present in your Schema Registry. Run this command from any node that can reach the Schema Registry:

curl http://<your-schema-registry-host>:<your-schema-registry-port>/subjects

Look for emailfilters-value (or emailfilters-key if you're using Avro for keys) in the returned list. If it's missing, that's the core issue.

2. Check Connector Converter Configuration

Double-check your Elasticsearch Sink connector's config to ensure:

• value.converter is set to io.confluent.connect.avro.AvroConverter (this is what triggers Schema Registry lookups)
• value.converter.schema.registry.url points to your active Schema Registry instance (no typos in IP/port)
• If you've customized the subject naming strategy via value.converter.subject.name.strategy, confirm the subject name format matches what's registered (or needs to be registered)

3. Register the Missing Schema (If Needed)

If the subject doesn't exist, you need to register the correct Avro schema for the emailfilters topic. If your producers are supposed to auto-register schemas, check that they're configured correctly with the Schema Registry URL. If you need to manually register the schema, use this command (replace the schema JSON with your actual record structure):

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
--data '{"schema": "{\"type\": \"record\", \"name\": \"EmailFilterRecord\", \"fields\": [{\"name\": \"filter_id\", \"type\": \"string\"}, {\"name\": \"email_pattern\", \"type\": \"string\"}]}"}' \
http://<your-schema-registry-host>:<your-schema-registry-port>/subjects/emailfilters-value/versions

4. Validate Network Connectivity to Schema Registry

From the Kafka Connect worker node (10.192.226.24), test that it can reach the Schema Registry:

curl http://<your-schema-registry-host>:<your-schema-registry-port>/

You should get a JSON response like {"schema_registry_version":"4.0.0"} (matching your Confluent version). If not, fix network rules or DNS issues blocking access.

5. Confirm Connector Topic Configuration

Make sure your connector's topics config is exactly emailfilters (no typos, extra spaces, or wrong topic names). A misspelled topic here would lead to the converter looking for a non-existent subject.

Verify the Fix #

After addressing the above, restart your connector and check its status:

./bin/confluent stop elasticsearch-sink && ./bin/confluent start elasticsearch-sink
./bin/confluent status elasticsearch-sink

Your task should now show a RUNNING state if the issue is resolved.

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