XML文件入Kafka Topic、REST API查询及发布异常问题咨询

Hey there! Let's walk through each of your questions with practical, actionable steps—since you already know your way around Avro and Kafka, this should align nicely with your existing workflow.

1. How to Publish XML Data to a Kafka Topic? #

You’ve got two solid approaches here, depending on whether you want to send raw XML or leverage your Avro expertise:

• Option 1: Send Raw XML Strings

  • First, split your XML file into individual records. Use libraries like Jackson XML or JAXB to parse the file and extract distinct nodes (e.g., each <member> entry in memberdetails.xml).
  • Use your existing Kafka Producer to send each parsed XML string as the value of a ProducerRecord. Assign a meaningful key (like a member ID or line number) to ensure consistent partitioning.
  • Quick code snippet for parsing and sending:

    // Using Jackson XML to parse XML nodes
    XmlMapper xmlMapper = new XmlMapper();
    File xmlFile = new File("testfolder/memberdetails.xml");
    MemberList memberList = xmlMapper.readValue(xmlFile, MemberList.class);
    
    for (Member member : memberList.getMembers()) {
        String xmlRecord = xmlMapper.writeValueAsString(member);
        // Use member ID as key for better consistency than line number
        producer.sendMessage(member.getId(), xmlRecord);
    }
    

• Option 2: Convert XML to Avro First

  • Map your XML structure to an Avro schema (define fields that match your XML nodes).
  • Parse XML into POJOs, then serialize those POJOs to Avro bytes using tools like the Avro Maven Plugin or Jackson Avro.
  • Send the Avro bytes as the value in your ProducerRecord (make sure to use KafkaAvroSerializer for the value serializer).

2. How to Access Kafka Data via REST API When the File Has Multiple Records? #

Kafka isn’t built for direct querying, so you’ll need to bridge the gap with one of these practical patterns:

• Pattern 1: Consume to a Queryable Store + Custom REST API

  • Set up a Kafka Consumer (or use Kafka Connect) to pull records from your topic and write them to a database or search engine like PostgreSQL, MySQL, or Elasticsearch.
  • Build a REST API (using Spring Boot, Flask, etc.) that queries this store to return data to clients. This is perfect for historical data and complex query needs.
  • Example flow: Kafka Topic → Kafka Connect JDBC Sink → PostgreSQL → Spring Boot REST API.

• Pattern 2: Kafka Streams + Interactive Queries

  • Use Kafka Streams to process your topic into a key-value state store, where each record is indexed by its key.
  • Enable Kafka Streams’ Interactive Queries feature, which lets you expose the state store directly via a REST endpoint. This is ideal for real-time, low-latency lookups.
  • Example: Create a KStream from your topic, convert it to a KTable for key-based persistence, then use the Streams API to fetch records by key via REST.

• Pattern 3: Use ksqlDB (Low-Code Option)

  • Define a stream in ksqlDB that maps to your Kafka topic.
  • Use ksqlDB’s built-in REST API to run SQL queries against the stream (e.g., SELECT * FROM member_stream WHERE id = '123';). This skips building a custom API entirely.

3. Why Is the Key NULL When Using kafka-console-producer? #

The default kafka-console-producer doesn’t handle keys unless you explicitly configure it to parse them. Here’s how to fix it:

• When starting the console producer, add these properties to enable key parsing:

  kafka-console-producer.sh --broker-list <your-broker-url>:9092 --topic <your-topic> \
    --property parse.key=true \
    --property key.separator=:
  

• Now, send records in the format key:value (e.g., 1:<member><id>1</id>...</member>). The producer will split the input at the separator and set the key correctly.

For your custom Java producer code: Double-check that the linenum variable isn’t empty or null when passed to sendMessage(). If you’re reading line numbers from the file, make sure your counter is incrementing correctly (e.g., use a loop variable instead of relying on BufferedReader’s getLineNumber() which can have edge cases).

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