[FIX] Empty old_product_id in Kafka reassign messages causes consumer failures

Problem

The listings consumer was failing to process deduplication (reassign) messages with the error:

Invalid ID format in reassign message. 
  listing_id=68b3074daf194ba27eb8fb0a, 
  old_product_id=,                          ← EMPTY!
  new_product_id=6892502da7e32ed5bc1268bf, 
  new_product_variant_id=68b3074baf194ba27eb8fade
Error: '' is not a valid ObjectId

Root Cause

In apply_db_ops.py, the generate_kafka_corrections function tried to derive old_product_id by matching model names via merge_map:

old_product_ids = set()
for merge_model, keep_model in merge_map.items():
    if _norm(keep_model) == _norm(new_model):  # ← This matching logic failed
        old_prod = products_by_model.get(_norm(merge_model))
        if old_prod:
            old_product_ids.add(old_prod._id)

old_product_id = next(iter(old_product_ids)) if old_product_ids else ""  # ← Empty!

When the model name matching failed (due to normalization differences), old_product_id defaulted to empty string "", which caused ObjectId("") to fail in the consumer.

Impact

• All deduplication messages published to Kafka had empty old_product_id
• Consumer logged errors and committed messages as processed (effectively losing them)
• Deduplications were not applied to MongoDB

Fix

Track old_product_id directly in listing updates instead of deriving it from model names:

1. build_ops_for_dedupes: Added _old_product_id: merge_id to each listing update
2. generate_kafka_corrections: Extract old_product_id from listing update metadata
3. Added validation to skip corrections without valid old_product_id

Files Changed

• data-projects/product_clustering/apply_db_ops.py

Related

• Consumer code: backend/data_processor/src/message_handlers/listings_handler.py:489
• The consumer's error handling commits failed messages, which should be improved separately