apache / apache/pulsar-client-go

Unexpected chunk messages are not redelivered

Open
#1,397 3 comments 0 reactions 0 assignees View on GitHub
Dominant language
Go
Stars
745
Forks
389
Avg merge
3d 20h
Merged PRs (30d)
3

Description

#### Expected behavior

When chunking is enabled, a golang consumer client code sometimes remove chunked msg context from it data through the below logic:
```
if ctx == nil || ctx.chunkedMsgBuffer == nil || chunkID != ctx.lastChunkedMsgID+1 {
lastChunkedMsgID := -1
totalChunks := -1
if ctx != nil {
lastChunkedMsgID = int(ctx.lastChunkedMsgID)
totalChunks = int(ctx.totalChunks)
ctx.chunkedMsgBuffer.Clear()
}
pc.log.Warnf(fmt.Sprintf(
"Received unexpected chunk messageId %s, last-chunk-id %d, chunkId = %d, total-chunks %d",
msgID.String(), lastChunkedMsgID, chunkID, totalChunks))
pc.chunkedMsgCtxMap.remove(uuid)
pc.availablePermits.inc()

return nil
}
```
If pulsar-client-go is removing such chunked message ctx's then there should be some information back to the broker for re-attempting this message again.

#### Actual behavior

When a consumer client discards this message, broker never retries to send the same message again and the msgbacklog ouput (through cli)

```./pulsar-admin topics partitioned-stats persistent://dummy/my_topic/my_namespace | grep msgBacklog``` being non-zero. Sometimes 5, 10 or even 1 message

#### Steps to reproduce

Enable producer and consumer side chunking with the following configs

Backend Consumer config

```
consumer:
topic: "my_topic"
namespace: "my_namespace"
subscriptionName: "my_subscription"
subscriptionType: "key_shared"
receiverQueueSize: 1000
enableAutoScaledReceiverQueueSizeLimit: "true"
nackRedeliveryDelay: 30s
maxPendingChunkMessages: 100
expiryTimeOfIncompleteChunkSec: 120
autoAckIncompleteChunkEnabled: "false"
```

Backend Producer config

```
producer:
topic: "my_topic"
pulsarClientIsolationEnabled: false
enabledChunking: "true"
compressionType: "lz4"
compressionLevel: "faster"
sendTimeout: 60s
disableBlockIfQueFull: true
maxPendingMessages: 1000
hashingScheme: "murmur3_32_hash"
sendAsync: true
```

Broker configuration
```
broker:
image: apachepulsar/pulsar:latest
container_name: broker
hostname: broker
restart: no
environment:
- metadataStoreUrl=zk:zookeeper:2181
- zookeeperServers=zookeeper:2181
- clusterName=cluster-a
- managedLedgerDefaultEnsembleSize=1
- managedLedgerDefaultWriteQuorum=1
- managedLedgerDefaultAckQuorum=1
- advertisedAddress=broker
- advertisedListeners=external:pulsar://broker:6650
- PULSAR_MEM=-Xms512m -Xmx512m -XX:MaxDirectMemorySize=2048m
- maxMessageSize=1242880
depends_on:
zookeeper:
condition: service_healthy
bookie:
condition: service_started
ports:
- "6650:6650"
- "8080:8080"
command: bash -c "bin/apply-config-from-env.py conf/broker.conf && exec bin/pulsar broker"
healthcheck:
test: [ "CMD", "curl", "-f", "http://localhost:8080/admin/v2/clusters" ]
interval: 5s
timeout: 10s
retries: 5
```

Ingestion payload size is 15MB.

Send this request for 300 seconds and the `unexpected chunk` message will come within 10-15 seconds on local machine.

Note: Currently, we are not hitting `go c.discardChunkIfExpire(uuid, true, c.pc.options.expireTimeOfIncompleteChunk)` and `go c.discardOldestChunkMessage(c.pc.options.autoAckIncompleteChunk)` condition for discard. The potential problem lies over there also as no such mechanism for broker to retry is made.

#### System configuration
**Pulsar version**: 3.3.2 for broker and 0.15.1 for pulsar-client-go

Can, we get some help here, as majority of our pipelines are currently impacted with this flow.

Contributor guide

Open the contributing guide

Research direction

Start with the chunk-message handling logic shown in the report, then reproduce with the listed producer, consumer, and broker settings using a 15MB payload. Trace the discard paths, including unexpected chunks and the noted expiry and oldest-message paths, and verify how redelivery is communicated. Done means discarded chunked messages can be retried by the broker and the partitioned-stats msgBacklog no longer remains non-zero for these messages.

Written by the indexing model from the issue text.

Assessment

Tech stack
go
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.