hailin
/
rwadurian
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(deploy): use kafka-console-producer for tombstone messages 

kafkacat/kcat not available in containers. Switch to kafka-console-producer
with null.marker property to send tombstone messages for Debezium offset deletion.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
hailin 2026-01-12 23:45:38 -08:00
parent fe2d4c3bcf
commit f84e8b4700
1 changed files with 16 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
backend/services/deploy-mining.sh
View File

@ -983,40 +983,28 @@ full_reset() {
local offset_key="[\"$connector\",{\"server\":\"$topic_prefix\"}]"
log_info "Sending tombstone for: $connector (key: $offset_key)"
# Try using kafkacat/kcat to send tombstone (NULL message)
# -Z flag enables NULL message delivery
# -K| sets key separator to |
# Send tombstone (NULL value) using kafka-console-producer
# Use null.marker to mark __NULL__ as NULL value
# Format: key\t__NULL__ with parse.key=true and null.marker=__NULL__
local tombstone_sent=false
# Try Debezium Connect container first (has kafkacat)
local connect_container="rwa-debezium-connect"
if docker ps --format '{{.Names}}' 2>/dev/null | grep -q "^${connect_container}$"; then
if docker exec "$connect_container" which kafkacat &>/dev/null; then
if echo "${offset_key}|" | docker exec -i "$connect_container" kafkacat -P -Z -b kafka:9092 -t "$offset_topic" -K \| 2>/dev/null; then
log_success "Sent tombstone via kafkacat (debezium-connect) for: $connector"
tombstone_sent=true
fi
fi
fi
# Try Kafka container if Debezium Connect didn't work
if [ "$tombstone_sent" = false ]; then
if docker exec "$KAFKA_CONTAINER" which kafkacat &>/dev/null; then
if echo "${offset_key}|" | docker exec -i "$KAFKA_CONTAINER" kafkacat -P -Z -b localhost:9092 -t "$offset_topic" -K \| 2>/dev/null; then
log_success "Sent tombstone via kafkacat (kafka) for: $connector"
tombstone_sent=true
fi
elif docker exec "$KAFKA_CONTAINER" which kcat &>/dev/null; then
if echo "${offset_key}|" | docker exec -i "$KAFKA_CONTAINER" kcat -P -Z -b localhost:9092 -t "$offset_topic" -K \| 2>/dev/null; then
log_success "Sent tombstone via kcat for: $connector"
tombstone_sent=true
fi
fi
# Use kafka-console-producer which is available in Kafka container
# --property parse.key=true: Enable key parsing
# --property key.separator=<TAB>: Use literal tab as key-value separator
# --property null.marker=__NULL__: Treat __NULL__ as null value (tombstone)
# Note: Must use printf to properly pass tab character through SSH/docker
if printf '%s\t%s\n' "$offset_key" "__NULL__" | docker exec -i "$KAFKA_CONTAINER" kafka-console-producer \
--bootstrap-server localhost:9092 \
--topic "$offset_topic" \
--property parse.key=true \
--property "key.separator= " \
--property "null.marker=__NULL__" 2>/dev/null; then
log_success "Sent tombstone via kafka-console-producer for: $connector"
tombstone_sent=true
fi
if [ "$tombstone_sent" = false ]; then
log_warn "Could not send tombstone for $connector"
log_warn "Neither kafkacat nor kcat available, offset may persist"
fi
done