Confluent Platform 6.1.1技术问题：KSQL表无法查询到数据求助

Let’s break down the possible issues and fixes for your scenario where the Kafka topic has data but your KSQL table returns no results:

1. Mismatch Between Message Value Structure and KSQL Table Schema #

Looking at your setup:

• Your Kafka messages have a value like AU,Australia (two comma-separated fields)
• Your table definition only includes one non-primary-key column (countryname)

When using VALUE_FORMAT='DELIMITED', KSQL expects the number of fields in the message value to exactly match the number of non-primary-key columns in your table. Since your value has 2 fields but your table only defines 1 non-key column, KSQL will fail to parse these messages and drop them silently.

Fix: #

Adjust your table schema to match the value structure. If you want to include both the duplicate country code from the value and the country name, update the table definition:

CREATE TABLE COUNTRY_TABLE (
  countrycode VARCHAR PRIMARY KEY,  -- Pulled from Kafka message key
  value_countrycode VARCHAR,        -- First field from message value
  countryname VARCHAR               -- Second field from message value
) WITH (
  KAFKA_TOPIC = 'COUNTRY-CSV',
  VALUE_FORMAT='DELIMITED',
  KEY_FORMAT='KAFKA'  -- Explicitly set to match your string keys
);

Or if you don’t need the duplicate country code from the value, modify your producer to send only the country name as the value (e.g., AU:Australia instead of AU:AU,Australia), then keep your original table definition.

2. auto.offset.reset Applied After Table Creation #

You set SET 'auto.offset.reset' = 'earliest'; after creating the table. KSQL uses the auto.offset.reset value at the time of table creation to determine where to start consuming the Kafka topic. If the default value was latest when you created the table, changing it later won’t affect the existing table’s consumer offset.

Fix: #

1. Drop the existing table:

   DROP TABLE COUNTRY_TABLE;
   

2. Set the offset reset property first:

   SET 'auto.offset.reset' = 'earliest';
   

3. Recreate the table with the corrected schema (from step 1).

3. Check KSQL Server Logs for Parsing Errors #

Silent failures in message deserialization are often logged in the KSQL server logs. Look for entries like:

Failed to deserialize value from topic COUNTRY-CSV: ...

These logs will tell you exactly why messages are being rejected (e.g., schema mismatch, invalid data format).

4. Verify Table Configuration with DESCRIBE EXTENDED #

Run this command in the KSQL CLI to confirm your table is pointing to the correct topic and using the right formats:

DESCRIBE EXTENDED COUNTRY_TABLE;

Check that:

• KAFKA_TOPIC is set to COUNTRY-CSV
• KEY_FORMAT matches your key serialization (you’re using string keys, so it should be KAFKA)
• VALUE_FORMAT is DELIMITED

5. Ensure the Kafka Topic’s Key Format Matches KSQL Expectations #

Your producer sends keys as strings (using parse.key=true and key.separator=:), which aligns with KSQL’s default KEY_FORMAT='KAFKA', but explicitly defining it in the table’s WITH clause avoids any ambiguity (as shown in the fix for step 1).

After applying these steps, re-run your query:

SELECT countrycode, countryname FROM COUNTRY_TABLE EMIT CHANGES;

You should now see the messages from your Kafka topic.

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