[Issue 8382][Pulsar Function] Enable e2e encryption for Pulsar Function

[nlu90]

(If this PR fixes a github issue, please add Fixes #<xyz>.)

Fixes #8382

Motivation

Add the e2e encryption support for Pulsar Functions

Modifications

• Add CryptoConfig the encapsulate all the crypto related configs set by user
• Add CryptoSpec to Function protobuf to container crypto information internally
• Add CryptoUtils to help create instance, convert between CryptoConfig and CryptoSpec
• Add crypto validation method in ValidatorUtils to ensure the provided CryptoKeyReader Class has a ctor with Map arg
• Updated the cli to allow user set crypto for consumer/producer when submitting the function
• Update PulsarSource, PulsarSink to use the crypto config if provided

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): Added pulsar-client-messagecrypto-bc into puslar-function instance
• The admin cli options: Added --producer-config to allow setting crypto related configs.

Documentation

• Does this pull request introduce a new feature? Yes
• If yes, how is the feature documented? not documented
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation: #8431

Now user can enable e2e encryption as follows:

./bin/pulsar-admin functions create --jar pulsar-functions/java-examples/target/pulsar-functions-api-examples.jar \
--classname org.apache.pulsar.functions.api.examples.ExclamationFunction  \
--inputs "persistent://public/default/input" --output persistent://public/default/output --producer-config '{"cryptoConfig": {"cryptoKeyReaderClassName": "org.apache.pulsar.functions.api.examples.RawFileKeyReader", "cryptoKeyReaderConfig": {"PUBLIC": "/Users/nlu/workspace/stream-native/pulsar/keys/test_ecdsa_pubkey.pem", "PRIVATE":"/Users/nlu/workspace/stream-native/pulsar/keys/test_ecdsa_privkey.pem"}, "producerCryptoFailureAction": "FAIL", "encryptionKeys":["myapp1"]}}'

./bin/pulsar-admin functions create --jar pulsar-functions/java-examples/target/pulsar-functions-api-examples.jar \
--classname org.apache.pulsar.functions.api.examples.ExclamationFunction  \
--input-specs '{"persistent://public/default/output": {"cryptoConfig": {"cryptoKeyReaderClassName": "org.apache.pulsar.functions.api.examples.RawFileKeyReader", "cryptoKeyReaderConfig": {"PUBLIC": "/Users/nlu/workspace/stream-native/pulsar/keys/test_ecdsa_pubkey.pem", "PRIVATE":"/Users/nlu/workspace/stream-native/pulsar/keys/test_ecdsa_privkey_2.pem"}, "consumerCryptoFailureAction": "FAIL"}}}' \
--output persistent://public/default/output-2 

One requirement is the provided cryptoKeyReaderClass must has a constructor with Map<String, Object> parameter for initialization. And the argument is provided via cryptoKeyReaderConfig in the cli

[wolfstudy]

Why the SEND command value is 10, Is there any special meaning here? We still reserve the middle value?

[nlu90]

Reserve some single digit values just in case there're new actions added for the consumer failure.

Logically it's more clear. FAIL or 0 is the default behavior; 1~9 is for consumer; 10 and above is for producer.

[codelipenghui]

ping @sijie @srkukarni Could you please help review this PR?

[sijie]

@nlu90 great contribution!

[Anonymitaet]

Doc has been added in https://github.com/apache/pulsar/pull/8648/files