[fix] DLQ to handle bytes key properly

[dlg99]

Motivation

DLQ producer implicitly converts byte[] key into a bas64-encoded string representing that byte array.

Modifications

Handling setting of keys that are hasBase64EncodedKey() differently.

Verifying this change

• Make sure that the change passes the CI checks.

This change added a unit test.

Does this pull request potentially affect one of the following parts:

NO

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: dlg99#19

[lhotari]

LGTM. Good catch!

[lhotari]

A message could also have a separate orderingKey. It is used as the secondary key in Key_Shared ordering:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java

Lines 1187 to 1196 in 15b88d2

	protected byte[] peekMessageKey(Message<?> msg) {
	byte[] key = NONE_KEY;
	if (msg.hasKey()) {
	key = msg.getKeyBytes();
	}
	if (msg.hasOrderingKey()) {
	key = msg.getOrderingKey();
	}
	return key;
	}

The handling for orderingKey is missing completely. I filed a separate issue #23173

[lhotari]

These code snippets helped understand this PR:
key and keyBytes in TypedMessageBuilderImpl:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java

Lines 124 to 150 in f3c177e

	@Override
	public TypedMessageBuilder<T> key(String key) {
	getKeyValueSchema().ifPresent(keyValueSchema -> checkArgument(
	keyValueSchema.getKeyValueEncodingType() != KeyValueEncodingType.SEPARATED,
	"This method is not allowed to set keys when in encoding type is SEPARATED"));
	if (key == null) {
	msgMetadata.setNullPartitionKey(true);
	return this;
	}
	msgMetadata.setPartitionKey(key);
	msgMetadata.setPartitionKeyB64Encoded(false);
	return this;
	}
	@Override
	public TypedMessageBuilder<T> keyBytes(byte[] key) {
	getKeyValueSchema().ifPresent(keyValueSchema -> checkArgument(
	keyValueSchema.getKeyValueEncodingType() != KeyValueEncodingType.SEPARATED,
	"This method is not allowed to set keys when in encoding type is SEPARATED"));
	if (key == null) {
	msgMetadata.setNullPartitionKey(true);
	return this;
	}
	msgMetadata.setPartitionKey(Base64.getEncoder().encodeToString(key));
	msgMetadata.setPartitionKeyB64Encoded(true);
	return this;
	}

getKey and getKeyBytes in MessageImpl:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java

Lines 613 to 636 in 8c50a6c

	@Override
	public String getKey() {
	if (msgMetadata.hasPartitionKey()) {
	return msgMetadata.getPartitionKey();
	} else {
	return null;
	}
	}
	@Override
	public boolean hasBase64EncodedKey() {
	return msgMetadata.isPartitionKeyB64Encoded();
	}
	@Override
	public byte[] getKeyBytes() {
	if (!msgMetadata.hasPartitionKey() || msgMetadata.isNullPartitionKey()) {
	return null;
	} else if (hasBase64EncodedKey()) {
	return Base64.getDecoder().decode(getKey());
	} else {
	return getKey().getBytes(UTF_8);
	}
	}

[lhotari]

DLQ producer implicitly converts byte[] key into a bas64-encoded string representing that byte array.

@dlg99 I guess the key bytes itself doesn't change, but the encoding changes?
What type of problems has this caused?

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 74.53%. Comparing base (bbc6224) to head (e167ec0).
Report is 530 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23172      +/-   ##
============================================
+ Coverage     73.57%   74.53%   +0.96%     
+ Complexity    32624     2755   -29869     
============================================
  Files          1877     1920      +43     
  Lines        139502   144468    +4966     
  Branches      15299    15804     +505     
============================================
+ Hits         102638   107679    +5041     
+ Misses        28908    28531     -377     
- Partials       7956     8258     +302     

Flag	Coverage Δ
inttests	27.87% <44.44%> (+3.29%)	⬆️
systests	24.78% <27.77%> (+0.46%)	⬆️
unittests	73.86% <100.00%> (+1.02%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	79.27% <100.00%> (+1.70%)	⬆️

... and 504 files with indirect coverage changes

[dlg99]

DLQ producer implicitly converts byte[] key into a bas64-encoded string representing that byte array.

@dlg99 I guess the key bytes itself doesn't change, but the encoding changes? What type of problems has this caused?

Correct, the hasBase64EncodedKey state is lost thus the key is not decoded into bytes.
It caused problems with KeyValue<AVRO, AVRO, SEPARATED> data sent to DLQ - data from the DLQ cannot be read (avro exception, wrong format) with expected schema; it can be read only using KeyValue<BYTES, AVRO, SEPARATED> and manually decoding and parsing the key. This happens because it goes through getKeyBytes() which relies on hasBase64EncodedKey() status to decide whether to to do the base64 decoding

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java

Lines 628 to 636 in 15b88d2

	public byte[] getKeyBytes() {
	if (!msgMetadata.hasPartitionKey() || msgMetadata.isNullPartitionKey()) {
	return null;
	} else if (hasBase64EncodedKey()) {
	return Base64.getDecoder().decode(getKey());
	} else {
	return getKey().getBytes(UTF_8);
	}
	}

[lhotari]

Correct, the hasBase64EncodedKey state is lost thus the key is not decoded into bytes. It caused problems with KeyValue<AVRO, AVRO, SEPARATED> data sent to DLQ - data from the DLQ cannot be read (avro exception, wrong format) with expected schema; it can be read only using KeyValue<BYTES, AVRO, SEPARATED> and manually decoding and parsing the key. This happens because it goes through getKeyBytes() which relies on hasBase64EncodedKey() status to decide whether to to do the base64 decoding

Thanks for sharing the context @dlg99