apache / apache/pulsar-client-go
Bug: Encrypted batch messages fail to deserialize when compression is disabled
- Dominant language
- Go
- Stars
- 745
- Forks
- 389
- Avg merge
- 3d 20h
- Merged PRs (30d)
- 3
Description
### Bug: Encrypted batch messages fail to deserialize when compression is disabled
#### Expected behavior
Encrypted batch messages with no compression should be consumed successfully, just like encrypted batch messages with compression (LZ4, ZSTD, etc).
#### Actual behavior
Consumer fails to deserialize encrypted batch messages when compression is not applied. This can occur when `CompressionType: NoCompression` is explicitly set, or when compression is conditionally skipped (e.g., for small batches where compression would not provide benefit).
The root cause is in `pulsar/consumer_partition.go` lines 1331-1340: the `reader.ResetBuffer()` call is only executed when `msgMeta.UncompressedSize != nil && *msgMeta.UncompressedSize > 0`.
#### Steps to reproduce
Failing test
```go
func TestBatchProducerConsumerRSAEncryptionWithoutCompression(t *testing.T) {
client, err := NewClient(ClientOptions{
URL: lookupURL,
})
assert.Nil(t, err)
defer client.Close()
topic := fmt.Sprintf("my-topic-enc-no-comp-%v", time.Now().Nanosecond())
cryptoConsumer, err := client.Subscribe(ConsumerOptions{
Topic: topic,
Decryption: &MessageDecryptionInfo{
KeyReader: crypto.NewFileKeyReader("crypto/testdata/pub_key_rsa.pem",
"crypto/testdata/pri_key_rsa.pem"),
},
SubscriptionName: "crypto-subscription",
Schema: NewStringSchema(nil),
})
assert.Nil(t, err)
normalConsumer, err := client.Subscribe(ConsumerOptions{
Topic: topic,
SubscriptionName: "normal-subscription",
Schema: NewStringSchema(nil),
})
assert.Nil(t, err)
batchSize := 2
cryptoProducer, err := client.CreateProducer(ProducerOptions{
Topic: topic,
Encryption: &ProducerEncryptionInfo{
KeyReader: crypto.NewFileKeyReader("crypto/testdata/pub_key_rsa.pem",
"crypto/testdata/pri_key_rsa.pem"),
Keys: []string{"client-rsa.pem"},
},
Schema: NewStringSchema(nil),
CompressionType: NoCompression,
DisableBatching: false,
BatchingMaxMessages: uint(batchSize),
})
assert.Nil(t, err)
msgFormat := "my-message-%v"
totalMessages := 10
ctx := context.Background()
for i := 0; i < totalMessages; i++ {
_, err := cryptoProducer.Send(ctx, &ProducerMessage{
Value: fmt.Sprintf(msgFormat, i),
})
assert.Nil(t, err)
}
// try to consume with normal consumer
normalConsumerCtx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
defer cancel()
msg, err := normalConsumer.Receive(normalConsumerCtx)
// msg should be null as the consumer will not be able to decrypt
assert.NotNil(t, err)
assert.Nil(t, msg)
// try to consume the message by crypto consumer
// consumer should be able to read all the messages
var actualMessage *string
for i := 0; i < totalMessages; i++ {
msg, err := cryptoConsumer.Receive(ctx)
assert.Nil(t, err)
expectedMsg := fmt.Sprintf(msgFormat, i)
err = msg.GetSchemaValue(&actualMessage)
assert.Nil(t, err)
assert.Equal(t, expectedMsg, *actualMessage)
cryptoConsumer.Ack(msg)
}
}
```
#### System configuration
- **Pulsar client**: v0.19.0, v0.20.0, v0.21.0, master
- **Go/OS**: any
Contributor guide
Research direction
Start in pulsar/consumer_partition.go around lines 1331-1340 and run TestBatchProducerConsumerRSAEncryptionWithoutCompression. Trace the reader.ResetBuffer() path for messages without UncompressedSize; done means encrypted batches with NoCompression are consumed successfully while the normal consumer still cannot decrypt them.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 85/100