apache / apache/pulsar-client-go

Bug: Encrypted batch messages fail to deserialize when compression is disabled

Open Beginner friendly
#1,527 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.