实时通过Kafka发布文件内容及崩溃重启时避免重复数据的方案咨询

Great question! Let's walk through the solutions step by step, covering Kafka's native features, custom strategies, and ready-to-use tools to avoid duplicate file content sends after a producer crash.

First, let's clarify what Kafka can do out of the box:

• Idempotent Producers: This is Kafka's built-in solution for preventing duplicate messages. When enabled, the producer assigns a unique PID (Producer ID) and sequence number to every message. The Kafka broker will automatically discard any duplicate messages with the same PID and sequence number, even if the producer retries or restarts.
  To enable this in Python (using libraries like confluent-kafka), add these configs:

  producer_conf = {
      'bootstrap.servers': 'your-broker:9092',
      'enable.idempotence': True,
      'acks': 'all',  # Required for idempotency
      'retries': 5    # Adjust based on your tolerance for retries
  }
  

  Note: This only prevents duplicate messages at the broker level—it doesn't track your progress in reading the source file. If your producer crashes mid-file-read, it won't automatically pick up where it left off; it might re-send chunks it already processed (though the broker will filter duplicates).

• Transactional Producers: If you're sending messages across multiple topics/partitions atomically, transactions ensure either all messages are committed or none are. For single-partition file publishing, idempotence is usually sufficient.

To pick up exactly where you left off in the file after a crash, you must track the file's read offset explicitly (e.g., how many bytes/lines you've successfully sent). Kafka doesn't natively track this file-specific state—this is up to your producer to manage.

Here's how to implement it properly:

1. Store the offset in a durable, atomic way:
   • Use a local state file (with atomic writes to avoid corruption on crash: write to a temp file first, then replace the main offset file).
   • Or store the offset in a dedicated Kafka topic (using a consumer group to manage commit state, leveraging Kafka's built-in offset persistence).
2. Only update the offset after successful message delivery:
   Don't update the offset until you get confirmation from the broker that the message was sent (via the delivery callback).

Your idea of "fetching published data and counting bytes to resume" is risky because:

• Consumer-side byte counts don't account for Kafka's message overhead (serialization, headers, etc.), so they won't match your file's exact read position.
• Consumer offsets can be delayed or inconsistent with your producer's progress, leading to either missed data or re-sends.

You don't have to build everything from scratch:

• Kafka Connect FileStreamSourceConnector: This official Kafka Connect plugin is designed exactly for this use case. It reads files line-by-line (or in chunks) and sends content to Kafka, automatically tracking the file's read offset (stored either locally or in Kafka's internal offset topic). It handles crash recovery seamlessly—just configure the file path, target topic, and offset storage, and it takes care of the rest.
• Python Reference Script: If you want to stick with Python, here's a simplified, robust script that tracks file offsets and uses idempotent production:

  import os
  from confluent_kafka import Producer
  
  # Configuration
  OFFSET_STORE = "file_read_offset.txt"
  TARGET_TOPIC = "file-content-topic"
  SOURCE_FILE = "/path/to/your/file.txt"
  CHUNK_SIZE = 4096  # Adjust based on your needs
  
  def load_last_offset():
      """Load the last successfully sent file offset"""
      if os.path.exists(OFFSET_STORE):
          with open(OFFSET_STORE, "r") as f:
              return int(f.read().strip())
      return 0
  
  def save_offset(offset):
      """Atomically save the current offset to avoid corruption"""
      temp_file = f"{OFFSET_STORE}.tmp"
      with open(temp_file, "w") as f:
          f.write(str(offset))
      os.replace(temp_file, OFFSET_STORE)
  
  def on_delivery(err, msg):
      """Callback to confirm message delivery and save offset"""
      if err:
          print(f"Failed to send message: {err}")
      else:
          # Extract the offset from message headers and save it
          current_offset = int(msg.headers()[0][1].decode())
          save_offset(current_offset)
  
  def main():
      producer = Producer({
          "bootstrap.servers": "localhost:9092",
          "enable.idempotence": True,
          "acks": "all"
      })
  
      start_offset = load_last_offset()
      with open(SOURCE_FILE, "rb") as f:
          f.seek(start_offset)
          while True:
              chunk = f.read(CHUNK_SIZE)
              if not chunk:
                  break
              # Attach current file offset as a message header
              producer.produce(
                  TARGET_TOPIC,
                  value=chunk,
                  headers=[("file-offset", str(f.tell()).encode())],
                  callback=on_delivery
              )
              producer.poll(0)  # Trigger callback processing
      producer.flush()  # Wait for all remaining messages to be sent
  
  if __name__ == "__main__":
      main()
  

• Use idempotent producers to prevent duplicate messages from reaching the broker, but pair it with explicit offset tracking to resume file reading exactly where you left off.
• Avoid relying on consumer-side metrics to track producer progress—it's unreliable and asynchronous.
• For minimal code overhead, use Kafka Connect's FileStreamSourceConnector—it's battle-tested and handles all edge cases for file-to-Kafka publishing.

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