Apache Pulsar: reproduce silent loss / stuck backlog producing into an empty fenced+CLOSED head ledger (BK-4812 family)

repro_empty_ledger_loss.sh

#!/usr/bin/env bash
#
# Reproduces silent message loss / stuck backlog when a producer writes the
# FIRST message into a managed-ledger whose empty head ledger has been fenced
# and CLOSED in metadata (length=0, lastEntryId=-1) by BookKeeper auto-recovery,
# while the broker still holds that ledger OPEN in memory.
#
# Root cause family: https://github.com/apache/bookkeeper/issues/4812
#
# Cluster: 1 ZooKeeper + 1 broker + 3 bookies, all on a user-defined Docker
# bridge with static IPs so a decommissioned bookie is replaced by a FRESH
# process at the SAME ip:port (fresh cookie, no in-memory fence state).
#
# Tested with image apachepulsar/pulsar:4.0.10
#
set -uo pipefail

IMG=apachepulsar/pulsar:4.0.10
NET=bkrepro
ZK_IP=172.30.0.2
BROKER_IP=172.30.0.3
declare -A BK=( [bk1]=172.30.0.11 [bk2]=172.30.0.12 [bk3]=172.30.0.13 )
TOPIC="persistent://public/default/empty-ledger-loss"
URL="pulsar://${BROKER_IP}:6650"
# Must be >= managedLedgerMaxLedgerRolloverTimeMinutes set on the broker below.
# The bug does not depend on the exact value; lower both to e.g. 2 for a faster run.
ROLL_MIN=15

say(){ echo -e "\n=== $* ==="; }

start_bookie(){ # name ip   -> FRESH ephemeral storage => fresh cookie
  local name="$1" ip="$2"
  docker rm -f "$name" >/dev/null 2>&1 || true
  docker run -d --name "$name" --network "$NET" --ip "$ip" \
    -e PULSAR_MEM="-Xms256m -Xmx256m -XX:MaxDirectMemorySize=256m" \
    -e clusterName=cl1 \
    -e metadataServiceUri="metadata-store:zk:${ZK_IP}:2181" \
    -e advertisedAddress="$ip" \
    -e useHostNameAsBookieID=false \
    -e autoRecoveryDaemonEnabled=true \
    -e openLedgerRereplicationGracePeriod=30000 \
    -e httpServerEnabled=true \
    -e PULSAR_PREFIX_httpServerPort=8000 \
    "$IMG" \
    bash -c "bin/apply-config-from-env.py conf/bookkeeper.conf && exec bin/pulsar bookie" >/dev/null
}

stats(){ docker exec broker bin/pulsar-admin topics stats-internal "$TOPIC" 2>/dev/null \
  | python3 -c "import sys,json;d=json.load(sys.stdin);print('  state',d['state'],'lastConfirmedEntry',d['lastConfirmedEntry'],'numberOfEntries',d['numberOfEntries'],'ledgers',[l['ledgerId'] for l in d['ledgers']])"; }
backlog(){ docker exec broker bin/pulsar-admin topics stats "$TOPIC" 2>/dev/null \
  | python3 -c "import sys,json;d=json.load(sys.stdin);s=d['subscriptions']['sub'];print('  msgBacklog',s['msgBacklog'])"; }
ledgercreated(){ docker exec broker bin/pulsar-admin topics stats-internal "$TOPIC" 2>/dev/null \
  | python3 -c "import sys,json,datetime;d=json.load(sys.stdin);t=d['lastLedgerCreatedTimestamp'];print(int(datetime.datetime.strptime(t,'%Y-%m-%dT%H:%M:%S.%fZ').replace(tzinfo=datetime.timezone.utc).timestamp()))"; }
zkmeta(){ docker exec bk2 bash -c "bin/bookkeeper shell ledgermetadata -ledgerid $1 2>/dev/null" 2>&1 \
  | grep -oE 'state=[A-Z]+, length=[0-9]+, lastEntryId=[0-9-]+|ensembles=\{[^}]*\}'; }
name_for_ip(){ for n in "${!BK[@]}"; do [ "${BK[$n]}" = "$1" ] && { echo "$n"; return; }; done; }

# wait until a bookie's ephemeral registration has expired from ZK after stop
dewait(){ for i in $(seq 1 40); do
  docker exec "$2" bash -c "bin/bookkeeper shell listbookies -rw 2>/dev/null" 2>&1 \
    | grep -oE 'BookieID:172.30.0.[0-9]+:3181' | grep -q "$1" && sleep 5 || { echo "    $1 deregistered"; return; }
done; }

decom(){ # ip runner  -> stop, wait deregister, decommission, FRESH restart at same ip
  local ip="$1" runner="$2" nm; nm=$(name_for_ip "$ip")
  echo "  -- decommission $nm ($ip) --"; docker stop "$nm" >/dev/null
  dewait "$ip:3181" "$runner"
  docker exec "$runner" bash -c "bin/bookkeeper shell decommissionbookie -bookieid $ip:3181 2>/dev/null" 2>&1 \
    | grep -iE 'properly replicated|deleted successfully' | grep -iv jar | sed 's/^/    /'
  start_bookie "$nm" "$ip"; sleep 8
}

############################################################
say "0. CLEANUP"
docker rm -f consumer broker zk bk1 bk2 bk3 >/dev/null 2>&1
docker network rm "$NET" >/dev/null 2>&1
docker network create --subnet 172.30.0.0/16 "$NET" >/dev/null

say "1. START ZooKeeper"
docker run -d --name zk --network "$NET" --ip "$ZK_IP" \
  -e PULSAR_MEM="-Xms256m -Xmx256m" "$IMG" \
  bash -c "bin/apply-config-from-env.py conf/zookeeper.conf && exec bin/pulsar zookeeper" >/dev/null
for i in $(seq 1 30); do docker exec zk bash -c "echo ruok | nc 127.0.0.1 2181" 2>/dev/null | grep -q imok && break; sleep 2; done

say "2. INITIALIZE cluster metadata"
docker run --rm --network "$NET" "$IMG" bin/pulsar initialize-cluster-metadata \
  --cluster cl1 \
  --metadata-store "zk:${ZK_IP}:2181" \
  --configuration-metadata-store "zk:${ZK_IP}:2181" \
  --web-service-url "http://${BROKER_IP}:8080" \
  --broker-service-url "pulsar://${BROKER_IP}:6650" >/dev/null 2>&1

say "3. START 3 FRESH bookies"
for n in bk1 bk2 bk3; do start_bookie "$n" "${BK[$n]}"; done
sleep 12

say "4. START broker (ensemble=2/2/2, maxRollover=${ROLL_MIN}m, min untouched)"
docker run -d --name broker --network "$NET" --ip "$BROKER_IP" \
  -e PULSAR_MEM="-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m" \
  -e PULSAR_PREFIX_clusterName=cl1 \
  -e PULSAR_PREFIX_metadataStoreUrl="zk:${ZK_IP}:2181" \
  -e PULSAR_PREFIX_configurationMetadataStoreUrl="zk:${ZK_IP}:2181" \
  -e PULSAR_PREFIX_bookkeeperMetadataServiceUri="metadata-store:zk:${ZK_IP}:2181" \
  -e PULSAR_PREFIX_advertisedAddress="$BROKER_IP" \
  -e PULSAR_PREFIX_managedLedgerDefaultEnsembleSize=2 \
  -e PULSAR_PREFIX_managedLedgerDefaultWriteQuorum=2 \
  -e PULSAR_PREFIX_managedLedgerDefaultAckQuorum=2 \
  -e PULSAR_PREFIX_managedLedgerMaxLedgerRolloverTimeMinutes="$ROLL_MIN" \
  -e PULSAR_PREFIX_brokerDeleteInactiveTopicsEnabled=false \
  -e PULSAR_PREFIX_systemTopicEnabled=false \
  -e PULSAR_PREFIX_topicLevelPoliciesEnabled=false \
  "$IMG" \
  bash -c "bin/apply-config-from-env.py conf/broker.conf && exec bin/pulsar broker" >/dev/null
for i in $(seq 1 40); do docker exec broker bin/pulsar-admin brokers healthcheck >/dev/null 2>&1 && break; sleep 3; done

############################################################
say "A. Subscribe a consumer => creates topic + EMPTY head ledger (NO message produced)"
docker run -d --name consumer --network "$NET" "$IMG" \
  bin/pulsar-client --url "$URL" consume -s sub -p Earliest -n 0 -t Shared "$TOPIC" >/dev/null
sleep 10
stats
CUR=$(docker exec broker bin/pulsar-admin topics stats-internal "$TOPIC" 2>/dev/null \
  | python3 -c "import sys,json;print(json.load(sys.stdin)['ledgers'][-1]['ledgerId'])")
ENS_IPS=$(docker exec bk2 bash -c "bin/bookkeeper shell ledgermetadata -ledgerid $CUR 2>/dev/null" 2>&1 \
  | grep -oE 'ensembles=\{[^}]*\}' | grep -oE '172.30.0.1[0-9]')
E1=$(echo "$ENS_IPS" | sed -n 1p); E2=$(echo "$ENS_IPS" | sed -n 2p)
SPARE=""; for ip in "${BK[@]}"; do echo "$ENS_IPS" | grep -q "$ip" || SPARE="$ip"; done
SP=$(name_for_ip "$SPARE")
echo "  EMPTY head ledgerId=$CUR  ensemble=[$E1,$E2]  spare/runner=$SPARE($SP)"
echo "  ZK head ledger: $(zkmeta "$CUR")"

say "B. Disconnect the consumer (stays offline through C/D/E)"
docker stop consumer >/dev/null; echo "  consumer stopped"

say "C. Decommission + FRESH-restart both bookies of the empty ledger's ensemble"
decom "$E1" "$SP"
decom "$E2" "$SP"
echo "  ZK head ledger $CUR now: $(zkmeta "$CUR")          <-- CLOSED, length=0, lastEntryId=-1"
echo "  broker still holds it OPEN in memory:"; stats

say "D. Wait >= managedLedgerMaxLedgerRolloverTimeMinutes(${ROLL_MIN}m) -- empty ledger is NOT auto-rolled"
CRE=$(ledgercreated); TARGET=$((CRE + ROLL_MIN*60 + 25))
echo "  waiting until $(date -u -r $TARGET +%H:%M:%SZ) ..."
while [ "$(date -u +%s)" -lt "$TARGET" ]; do sleep 15; done
echo "  state just before produce (still the same empty ledger $CUR):"; stats

say "E. Produce the FIRST message (consumer still OFFLINE) -> lands in the fenced/CLOSED ledger"
docker run --rm --network "$NET" "$IMG" \
  bin/pulsar-client --url "$URL" produce "$TOPIC" -m msg1 -n 1 2>&1 \
  | grep -iE 'successfully produced|ERROR' | grep -iv jar | sed 's/^/  /'
sleep 8
echo "  broker after produce:"; stats; backlog
echo "  ZK durable head ledger $CUR: $(zkmeta "$CUR")"

say "F. Reconnect the consumer -- can it receive msg1?  (EXPECT: NO; backlog stuck at 1)"
docker rm -f consumer >/dev/null 2>&1
docker run -d --name consumer --network "$NET" "$IMG" \
  bin/pulsar-client --url "$URL" consume -s sub -p Earliest -n 0 -t Shared "$TOPIC" >/dev/null
sleep 14
echo "  consumer received: [$(docker logs consumer 2>&1 | grep -oE 'content:[a-z0-9]+' | tr '\n' ' ')]"
backlog

say "G. Force reconcile (unload) -- phantom entry discarded, backlog -> 0, msg1 SILENTLY LOST"
docker stop consumer >/dev/null 2>&1
docker exec broker bin/pulsar-admin topics unload "$TOPIC" >/dev/null 2>&1; sleep 8
echo "  durable view after reconcile:"; stats; backlog
docker rm -f consumer >/dev/null 2>&1
docker run -d --name consumer --network "$NET" "$IMG" \
  bin/pulsar-client --url "$URL" consume -s sub -p Earliest -n 0 -t Shared "$TOPIC" >/dev/null
sleep 12
echo "  consumer received post-reconcile: [$(docker logs consumer 2>&1 | grep -oE 'content:[a-z0-9]+' | tr '\n' ' ')]"
backlog

say "DONE.  Cleanup: docker rm -f consumer broker zk bk1 bk2 bk3 && docker network rm $NET"