Can't configure a cryptoKeyReader via ReaderBuilder.loadConf

[objecttrouve]

Describe the bug

I'm trying to configure a CryptoKeyReader with a FlinkPulsarSource.

The Pulsar docs suggest you can configure such a crypto key reader by passing a value for property called cryptoKeyReader.
Here's where those properties are used (on creating the Flink Pulsar source).

The evaluating code tries to deserialize a config object from the property map.
But what would be the value in such a map? It's not the class name, as I naively assumed.

If I understand the code correctly, the value doesn't even matter, because the object mapper tries to instantiate an interface. Resulting in the following error:

com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of `org.apache.pulsar.client.api.CryptoKeyReader` (no Creators, like default constructor, exist): abstract types either need to be mapped to concrete types, have custom deserializer, or contain additional type information

I also asked about this in the flink-connector channel on Slack. But unless I'm missing something, I think it's a bug in Pulsar. Or at least a misleading documentation. Raising this issue to clarify.

(If the calling code was under my control, I could, of course, just use the setter for the crypto key reader. I might do a PR about it here or in the pulsar-flink project. However, I want to clarify the intended usage first. Moreover, I have doubts that my use case is so useful, since most of the time you'll want to configure the crypto key reader.)

To Reproduce

Here's a unit test illustrating what I was trying to do:

@SuppressWarnings("unused")
public static class MyCryptoKeyReader implements CryptoKeyReader {

	@Override
	public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> metadata) {
		return null;
	}

	@Override
	public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> metadata) {
		return null;
	}
}

	@Ignore("Illustrates wrong usage.")
	@Test
	public void readerBuilderLoadConfWithCryptoKeyReaderClassValueTest() throws Exception {

		@Cleanup
		PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();

		Map<String, Object> config = new HashMap<>();

		/*
		* This is what I first assumed.
		* This ISN'T how it works!
		* The `loadConf` method expects a _serialized_ `CryptoKeyReader` at this point.
		*/

		config.put("cryptoKeyReader", "org.apache.pulsar.client.impl.BuildersTest$MyCryptoKeyReader");

		ReaderBuilderImpl<byte[]> builder = (ReaderBuilderImpl<byte[]>) client.newReader()
		.loadConf(config);

		Class<?> clazz = builder.getClass();
		Field conf = clazz.getDeclaredField("conf");
		conf.setAccessible(true);
		Object obj = conf.get(builder);
		assertTrue(obj instanceof ReaderConfigurationData);
		//noinspection rawtypes
		assertTrue(((ReaderConfigurationData)obj).getCryptoKeyReader() instanceof MyCryptoKeyReader);

}

It reproduces the error mentioned above.

Expected behavior

I inspected a property where I could pass a class name of a CryptoKeyReader implementation with a no-arg constructor, which would be instantiated on loading the config.

Desktop (please complete the following information):

uname -a
Linux otto 5.4.0-80-generic #90-Ubuntu SMP Fri Jul 9 22:49:44 UTC 2021 x86_64 x86_64 x86_64 GNU/Linux

Additional context

Same question in the Pulsar Slack.

Issue 7579 is similar.

[objecttrouve]

Created #386 in the streamnative/pulsar-flink project.

[codelipenghui]

The issue had no activity for 30 days, mark with Stale label.

[tisonkun]

Closed as stale. Please create a new issue if it's still relevant to the maintained versions.