#386: Implemented passing of `CryptoKeyReader`. by objecttrouve · Pull Request #387 · streamnative/pulsar-flink

objecttrouve · 2021-08-08T13:23:44Z

Implemented passing a CryptoKeyReader (and encryption keys) to FlinkPulsarSource and FlinkPulsarSink, as requested here.

Used builder pattern for easy extensibility without breaking or excessively overloading public c'tors.

Added test. (Maybe it can be moved to one of the other tests to avoid overhead.)

As requested [here](streamnative#386), it's now possible to pass a `CryptoKeyReader` (and encryption keys) to `FlinkPulsarSource` and `FlinkPulsarSink`. Used builder pattern for easy extensibility without breaking or excessively overloading public c'tors. Added integration test. (Maybe it can be moved to one of the other tests to avoid overhead.)

nlu90 · 2021-08-10T18:15:49Z

@jianyun8023 @syhily could you help take a look?

syhily · 2021-08-27T15:06:42Z

Hi @objecttrouve, tks for your contribution. Using a builder pattern on internal Fetcher don't sounds like a Good design.
Add builder pattern for public source & sink is accepted, by we may need extra validation on final build method.

The detailed review advice would be given in weekend.

syhily · 2021-08-27T15:09:01Z

...nk-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSink.java

+
    private final PulsarSerializationSchema<T> serializationSchema;

+    public FlinkPulsarSink(final Builder<T> builder) {


I don't think this should be public for a Builder pattern.

Absolutely. Must have missed it. Will fix.

Made c'tors private.

syhily · 2021-08-27T15:10:16Z

...nk-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSink.java


+    public FlinkPulsarSink(final Builder<T> builder) {
+        super(
+            new FlinkPulsarSinkBase.Config<T>()


This could be boilerplate. Why not simply use ctor? Since it's a builder pattern. The end user would never touch this ctor.

The end user would never touch this ctor.

I wasn't sure about which classes an end user could touch. Therefore, I tried hard not to change the public c'tors plus not to add more than absolutely necessary. But if the class is just for internal use, right, it's unnecessary boilerplate. Will change it.

Back to the old c'tor.

syhily · 2021-08-27T15:12:09Z

...nk-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSink.java

-
-        super(adminUrl, defaultTopicName, clientConf, properties, serializationSchema, messageRouter, semantic);
-        this.serializationSchema = serializationSchema;
+        this(new Builder<T>()


Refactor old ctor doen't look like a good choice.

The goal was basically to delegate as much as possible to a single c'tor that contains the interesting logic, instead of repeating variants of the logic inside the c'tor all over the place.
But to be honest, I don't mind very much. So I'll just revert to the old c'tor.

Back to the old c'tor.

syhily · 2021-08-27T15:14:29Z

...nk-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSink.java

+            if (clientConf == null){
+                throw new IllegalStateException("Client conf must be set.");
+            }
+            if ((cryptoKeyReader == null) != (encryptionKeys.isEmpty())){


cryptoKeyReader requires extra serialization check. Use Preconditions.checkState(InstantiationUtil.isSerializable(cryptoKeyReader))

OK, will do.

Added the check.

syhily · 2021-08-27T15:15:11Z

...onnector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSinkBase.java

 @Slf4j
 abstract class FlinkPulsarSinkBase<T> extends TwoPhaseCommitSinkFunction<T, FlinkPulsarSinkBase.PulsarTransactionState<T>, Void> implements CheckpointedFunction {

+    public static class Config<T> {


Why we need builder for this base class?

Much the same reasoning as implemented throughout. A builder or config object makes it easier to add parameters in the future, without breaking an API or resulting in more and more overloaded c'tors. I have to admit I'm not sure which classes exactly were intended strictly for private purposes, so I just applied the pattern all over the place.
(Knowing that builders and config objects imply a lot of boilerplate.)

If overloading c'tors or changing the API of the class is OK, I'm completely fine with it though. So I'll change this piece, too.

Back to the old c'tor.

syhily · 2021-08-27T15:15:47Z

...onnector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSinkBase.java

-            PulsarSerializationSchema<T> serializationSchema,
-            MessageRouter messageRouter,
-            PulsarSinkSemantic semantic) {
+    public FlinkPulsarSinkBase(final Config<T> config) {


I think we may just keep this ctor.

Much the same as above, I'll change it.

Back to the old c'tor.

syhily · 2021-08-27T15:18:12Z

...-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSource.java

+            if (clientConf == null){
+                throw new IllegalStateException("Client conf mustn't be null. Either provide a client conf or a service URL plus properties.");
+            }
+            return new FlinkPulsarSource<>(this);


This cryptoKeyReader doesn't require encryptionKeys here. Right?

We still need to check the serialization for encryptionKeys here.

No, I think it doesn't require the encryptionKeys. But I'll double check.
OK, will check serialization.

In the source, no encryption keys are required. Added serialization check.

syhily · 2021-08-27T15:19:38Z

...-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSource.java

-                streamingRuntime.getMetricGroup().addGroup(PULSAR_SOURCE_METRICS_GROUP),
-                useMetrics
-        );
+        return new PulsarFetcher.Builder()


Since the PulsarFetcher is an internal API. Plz just keep the ctor.

Back to old c'tor.

objecttrouve · 2021-08-28T10:40:44Z

@syhily, thanks for the feedback!

I'll make the requested changes in my next free time slot.

Turned out builder pattern wasn't necessary here. There's one path now for the CryptoKeyReader (plus the encryption keys) and possibly others where it's null (and the keys are empty).

…yption keys.

objecttrouve · 2021-09-15T15:13:43Z

@syhily, I made the changes as I understood your requests. Please let me know if there's anything else. Also, if I should squash. (Left multiple commits for better relatability to your remarks. For the time being.)

syhily · 2021-09-15T15:50:52Z

Thanks for your contribution. The PR looks reasonable on my side. I will merge this on the weekend.

objecttrouve · 2021-09-21T18:22:25Z

Thanks for your contribution. The PR looks reasonable on my side. I will merge this on the weekend.

Thanks @syhily!

objecttrouve requested review from a team, jianyun8023 and syhily as code owners August 8, 2021 13:23

syhily suggested changes Aug 27, 2021

View reviewed changes

objecttrouve added 3 commits September 15, 2021 15:31

streamnative#386: Made c'tors private.

05e3b78

streamnative#386: Restored private API c'tors.

50b9e9b

Turned out builder pattern wasn't necessary here. There's one path now for the CryptoKeyReader (plus the encryption keys) and possibly others where it's null (and the keys are empty).

streamnative#386: Check serializability of crypto key reader and encr…

a0b27ca

…yption keys.

objecttrouve mentioned this pull request Oct 18, 2021

[FEATURE]: Configure CryptoKeyReader in FlinkPulsarSource and FlinkPulsarSink #386

Open


		private final PulsarSerializationSchema<T> serializationSchema;

		public FlinkPulsarSink(final Builder<T> builder) {

Conversation

objecttrouve commented Aug 8, 2021

Uh oh!

nlu90 commented Aug 10, 2021

Uh oh!

syhily commented Aug 27, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

objecttrouve Aug 28, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

objecttrouve commented Aug 28, 2021

Uh oh!

objecttrouve commented Sep 15, 2021

Uh oh!

syhily commented Sep 15, 2021

Uh oh!

objecttrouve commented Sep 21, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

objecttrouve Aug 28, 2021 •

edited

Loading