@lhotari Hi, lari. I've updated the test, PTAL

Codecov Report

❌ Patch coverage is 81.81818% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.23%. Comparing base (e5e7981) to head (7617179).
⚠️ Report is 2 commits behind head on master.

@@             Coverage Diff              @@
##             master   #24663      +/-   ##
============================================
+ Coverage     74.01%   74.23%   +0.22%     
+ Complexity    33224    33209      -15     
============================================
  Files          1858     1885      +27     
  Lines        146500   146949     +449     
  Branches      16880    16927      +47     
============================================
+ Hits         108425   109092     +667     
+ Misses        29394    29158     -236     
- Partials       8681     8699      +18     

I ran some manual verification and things seem to work in an expected way. The message will get produced to the DLQ topic with the original schema version. It seems to be a property of the AUTO_PRODUCE_BYTES schema or message.getReaderSchema() that it's referring to the wrong schema when it attempts to run the validate. The solution in this PR seems to be a reasonable workaround to address this issue until we find a better solution.

One possible reason might be that getReaderSchema() might only be accurate in AUTO_CONSUME mode:

I think I found an answer to the wrong schema instance when using AUTO_PRODUCE_BYTES with message.getReaderSchema().get() as the schema.

For example, for Avro messages, this maps finally here:

It will always use the original Schema instance for reading and never refresh the "readerSchema" from the schema registry. This is why the mandatory validation before serializing the message to DLQ fails. Considering this fact, I think that that AutoProduceBytesSchema.setRequireSchemaValidation(false) is a very reasonable way to workaround this issue for the DLQ use case.

I think I found an answer to the wrong schema instance when using AUTO_PRODUCE_BYTES with message.getReaderSchema().get() as the schema.

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

Lines 395 to 413 in 1220951

	public Optional<Schema<?>> getReaderSchema() {
	ensureSchemaIsLoaded();
	if (schema == null) {
	return Optional.empty();
	}
	byte[] schemaVersion = getSchemaVersion();
	if (schemaVersion == null) {
	return Optional.of(schema);
	}
	if (schema instanceof AutoConsumeSchema) {
	return Optional.of(((AutoConsumeSchema) schema)
	.atSchemaVersion(schemaVersion));
	} else if (schema instanceof AbstractSchema) {
	return Optional.of(((AbstractSchema<?>) schema)
	.atSchemaVersion(schemaVersion));
	} else {
	return Optional.of(schema);
	}
	}

For example, for Avro messages, this maps finally here:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/MultiVersionAvroReader.java

Lines 53 to 54 in 3b7bef1

	return new AvroReader<>(parseAvroSchema(schemaInfo.getSchemaDefinition()),
	readerSchema, pojoClassLoader, jsr310ConversionEnabled);

It will always use the original Schema instance for reading and never refresh the "readerSchema" from the schema registry. This is why the mandatory validation before serializing the message to DLQ fails. Considering this fact, I think that that AutoProduceBytesSchema.setRequireSchemaValidation(false) is a very reasonable way to workaround this issue for the DLQ use case.

@lhotari thank you so much for sharing this. this is very helpful and it's clear to me now. BTW, this solution is more reasonable than my initial attempt.

There's one remaining detail here. This solution will only work if SchemaCompatibilityStrategy is configured as ALWAYS_COMPATIBLE for the DLQ topic in the topic policy or in the namespace level. While testing this in action, it appeared that updating the SchemaCompatibilityStrategy changes at the namespace level aren't effective after a topic has been loaded. (example: this commit lhotari/pulsar-dlq-schema-issue-example@5799194 "fixed" the test app that was shared by @tonisojandu)