PIP-85: [pulsar-io] pass pulsar client via context to connector #11056

nlu90 · 2021-06-23T22:17:14Z

Motivation

Modifications

Expose PulsarClient via BaseContext, and allow connectors to use the inherited pulsar client from function worker to produce/consume messages.

Verifying this change

Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as:

PulsarOffsetBackingStoreTest
KafkaConnectSourceTest
KafkaConnectSinkTest

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

The public API: SourceContext and SinkContext need to implement the getPulsarClient method

sijie

Overall looks good.

eolivelli

Thank you very much for bringing this patch to the community.

There has been a long discussion recently about adding this feature and in general to add support for accessing the Pulsar Client I to Pulsar IO.

There are some security concerns because the client we be able to leverage all of the resources of the client.

Also we have to ensure that the API that we expose will work correctly.

This patch also touches the Kafka integration part, probably it is better to split the patch into two parts.

I appreciate your initiative but this is kind of an hot topic in the community and it deserves a little bit of discussion.

During the past months we added new API like this one (see the Pulsar Admin API support) and it was a pain.

There are pending works about allowing to use the Pulsar Client API inside functions and IO connectors because we have classpath problems.

My suggestion here is to:

Start a discussion on the dev@
define clearly how functions and IO connectors must refer to the Pulsar API, for instance currently many connectors must bundle Pulsar client jars inside the nar file
add integration tests that cover the list of supported API in the client (do we have to fully support the client API?)

@merlimat @jerrypeng @codelipenghui @lhotari @dlg99 please take a look as you are interested in this topic.

I am overall okay in adding this feature but this time we have to think more about the design of such feature

...a-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSinkTest.java

dlg99

It is a useful thing to have, some minor nits to clean up.

I share @eolivelli 's concerns and looking forward for the consistent way forward for the Client/Admin in the context API that is agreed on by the community.

...nnect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarOffsetBackingStore.java

pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/BaseContext.java

dlg99 · 2021-06-24T19:58:37Z

also it broke the tests

Error:  Tests run: 37, Failures: 1, Errors: 0, Skipped: 27, Time elapsed: 79.186 s <<< FAILURE! - in org.apache.pulsar.io.kafka.connect.KafkaConnectSinkTest
Error:  offsetTest(org.apache.pulsar.io.kafka.connect.KafkaConnectSinkTest)  Time elapsed: 2.084 s  <<< FAILURE!
java.lang.RuntimeException: Failed to create pulsar client to cluster at http://localhost:43199
	at org.apache.pulsar.io.kafka.connect.PulsarOffsetBackingStore.start(PulsarOffsetBackingStore.java:159)
	at org.apache.pulsar.io.kafka.connect.PulsarKafkaSinkTaskContext.<init>(PulsarKafkaSinkTaskContext.java:73)
	at org.apache.pulsar.io.kafka.connect.KafkaConnectSink.open(KafkaConnectSink.java:163)
	at org.apache.pulsar.io.kafka.connect.KafkaConnectSinkTest.offsetTest(KafkaConnectSinkTest.java:439)
...
Caused by: org.apache.pulsar.client.api.PulsarClientException$AlreadyClosedException: Client already closed : state = Closed
	at org.apache.pulsar.client.api.PulsarClientException.unwrap(PulsarClientException.java:975)
	at org.apache.pulsar.client.impl.ProducerBuilderImpl.create(ProducerBuilderImpl.java:95)
	at org.apache.pulsar.io.kafka.connect.PulsarOffsetBackingStore.start(PulsarOffsetBackingStore.java:147)

nlu90

Thank you for your active review. I'll put this PR into review once all issues and failed tests are fixed.

...a-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSinkTest.java

...nnect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarOffsetBackingStore.java

sijie · 2021-06-25T01:16:26Z

@eolivelli

There are some security concerns because the client we be able to leverage all of the resources of the client.

What are the security concerns? I think we have discussed that in one of the community meetings talking about the PulsarAdmin interface. The function uses the credentials/token to access the Pulsar resources. The credentials already limit the access on what resources that the function can access. I don't see why and how expose PulsarAdmin or PulsarClient can introduce security issues.

Also we have to ensure that the API that we expose will work correctly.

This patch also touches the Kafka integration part, probably it is better to split the patch into two parts.

The main reason for this change is to solve the problem of the Debezium connector. A Debezium connector is a wrapper over Kafka connector. I don't see why we need to split that into two parts.

we added new API like this one (see the Pulsar Admin API support) and it was a pain.
There are pending works about allowing to use of the Pulsar Client API inside functions and IO connectors because we have classpath problems.

We have fixed the class loading problem. If we don't encounter class loading problems, I don't know why this would be a concern.

define clearly how functions and IO connectors must refer to the Pulsar API, for instance currently many connectors must bundle Pulsar client jars inside the nar file

I don't see how is that related to this pull request. These are two different issues. Please don't couple them together.

add integration tests that cover the list of supported API in the client (do we have to fully support the client API?)

This change is used for replacing the Pulsar client in the Pulsar offset store, which already points to the same Pulsar cluster. So it is already being tested as part of debezium connector integration tests.

dlg99

LGTM

jerrypeng · 2021-06-25T16:50:45Z

@eolivelli the auth data the pulsar client that is instantiated in a pulsar instance has should be properly scoped. Usually it is inherits the credentials of the the source/sink/function submitter. Thus, the instance will only be able to perform the operations that it is authorized to perform.

dlg99 · 2021-06-27T17:53:06Z

@sijie see #11099 regarding

it is already being tested as part of debezium connector integration tests.

eolivelli · 2021-06-28T06:13:22Z

@jerrypeng thanks for your clarification.

@sijie
I am totally fine with adding this feature, and also I am very happy to see the Debezium problem fixed.

But my point is that here we are adding again a new powerful API to the Pulsar Functions/IO system.
So this kind of changes in my humble option MUST be discussed properly on the dev@ mailing list.
This way the project/community will be aware of what's going on and have the chance to participate into the discussion
for a new API.
A PR is not good for such high level discussions, because not everyone is following the huge quantity of messages due to PR.

We said many times during the past months (cc @merlimat) that we should use better the tool of the PIPs and we should add new APIs more carefully, being sure that the community is aware of what is happening.

eolivelli · 2021-06-28T06:15:59Z

@sijie the "classpath problem" is not solved.
Here it is @merlimat 's patch, that is still not committed.
#10922

With this change we are going to add more access to the Pulsar API, but not every API that is needed to leverage the Pulsar client is working.

nlu90 · 2021-06-28T20:05:00Z

Hi, seems the CI check failed for some weird reason.

Can anyone re-run all the test?

eolivelli · 2021-06-28T22:26:35Z

/pulsarbot rerun-failure-checks

sijie · 2021-06-28T23:18:05Z

So this kind of changes in my humble option MUST be discussed properly on the dev@ mailing list.

Agree on having a PIP. but it doesn't mean that we shouldn't include the Pulsar client API. Class loading is a different problem that we need to solve. We don't necessarily need to couple different issues into one discussion.

@nlu90 Can you send out a PIP to dev@ mailing list for this new API?

eolivelli · 2021-06-29T05:20:17Z

So this kind of changes in my humble option MUST be discussed properly on the dev@ mailing list.

Agree on having a PIP. but it doesn't mean that we shouldn't include the Pulsar client API. Class loading is a different problem that we need to solve. We don't necessarily need to couple different issues into one discussion.

Agreed. Thanks

freeznet · 2021-06-29T05:56:34Z

/pulsarbot run-failure-checks

nlu90 · 2021-06-30T19:57:16Z

So this kind of changes in my humble option MUST be discussed properly on the dev@ mailing list.

Agree on having a PIP. but it doesn't mean that we shouldn't include the Pulsar client API. Class loading is a different problem that we need to solve. We don't necessarily need to couple different issues into one discussion.

@nlu90 Can you send out a PIP to dev@ mailing list for this new API?

yeah, I just sent the PIP to dev mailing list.

sijie · 2021-06-30T22:16:31Z

@eolivelli Can you please review and unblock this PR?

eolivelli · 2021-07-01T06:00:13Z

for reference
https://github.com/apache/pulsar/wiki/PIP-85:-Expose-Pulsar-Client-via-Function-Connector-BaseContext

eolivelli

Thank you very much @nlu90

I have restarted the failed CI job, we can merge as soon as CI is green.

nlu90 · 2021-07-02T06:33:23Z

/pulsarbot run-failure-checks

nlu90 · 2021-07-02T16:53:03Z

/pulsarbot run-failure-checks

…y database (#11251) ### Motivation The Debezium requires pulsar a service URL for history database usage. In #11056 , the `service.url` field from `PulsarKafkaWorkerConfig` is no longer available. And the value is also deleted from multiple yaml config files in this [commit](3ce24c9). This causes the integration test for Debezium connector to fail. Based on the Debezium [paradigm](https://debezium.io/documentation/reference/1.5/connectors/mysql.html#debezium-mysql-connector-database-history-configuration-properties), all configurations should be passed as strings. There's no easy way to inject a PulsarClient via configuration. We need to ask user to provide the pulsar url explicitly and probably auth info also. ### Modifications 1. Make the `database.history.pulsar.service.url` field required 2. Add the config value back to example yaml files 3. Update the integration test config ### Verifying this change - [ ] Make sure that the change passes the CI checks.

…he#11056) ### Motivation Fixes apache#8668 ### Modifications Expose `PulsarClient` via `BaseContext`, and allow connectors to use the inherited pulsar client from function worker to produce/consume messages. ### Verifying this change - [ ] Make sure that the change passes the CI checks. This change is already covered by existing tests, such as: - PulsarOffsetBackingStoreTest - KafkaConnectSourceTest - KafkaConnectSinkTest ### Does this pull request potentially affect one of the following parts: - The public API: `SourceContext` and `SinkContext` need to implement the `getPulsarClient` method

…y database (apache#11251) ### Motivation The Debezium requires pulsar a service URL for history database usage. In apache#11056 , the `service.url` field from `PulsarKafkaWorkerConfig` is no longer available. And the value is also deleted from multiple yaml config files in this [commit](apache@3ce24c9). This causes the integration test for Debezium connector to fail. Based on the Debezium [paradigm](https://debezium.io/documentation/reference/1.5/connectors/mysql.html#debezium-mysql-connector-database-history-configuration-properties), all configurations should be passed as strings. There's no easy way to inject a PulsarClient via configuration. We need to ask user to provide the pulsar url explicitly and probably auth info also. ### Modifications 1. Make the `database.history.pulsar.service.url` field required 2. Add the config value back to example yaml files 3. Update the integration test config ### Verifying this change - [ ] Make sure that the change passes the CI checks.

…he#11056) ### Motivation Fixes apache#8668 ### Modifications Expose `PulsarClient` via `BaseContext`, and allow connectors to use the inherited pulsar client from function worker to produce/consume messages. ### Verifying this change - [ ] Make sure that the change passes the CI checks. This change is already covered by existing tests, such as: - PulsarOffsetBackingStoreTest - KafkaConnectSourceTest - KafkaConnectSinkTest ### Does this pull request potentially affect one of the following parts: - The public API: `SourceContext` and `SinkContext` need to implement the `getPulsarClient` method

…y database (apache#11251) ### Motivation The Debezium requires pulsar a service URL for history database usage. In apache#11056 , the `service.url` field from `PulsarKafkaWorkerConfig` is no longer available. And the value is also deleted from multiple yaml config files in this [commit](apache@3ce24c9). This causes the integration test for Debezium connector to fail. Based on the Debezium [paradigm](https://debezium.io/documentation/reference/1.5/connectors/mysql.html#debezium-mysql-connector-database-history-configuration-properties), all configurations should be passed as strings. There's no easy way to inject a PulsarClient via configuration. We need to ask user to provide the pulsar url explicitly and probably auth info also. ### Modifications 1. Make the `database.history.pulsar.service.url` field required 2. Add the config value back to example yaml files 3. Update the integration test config ### Verifying this change - [ ] Make sure that the change passes the CI checks.

Cherry pick #11293 ## Motivation The Debezium requires pulsar a service URL for history database usage. In #11056 , the service.url field from PulsarKafkaWorkerConfig is no longer available. And the value is also deleted from multiple yaml config files in this commit. This causes the integration test for Debezium connector to fail. Based on the Debezium paradigm, all configurations should be passed as strings. There's no easy way to inject a PulsarClient via configuration. We need to ask user to provide the pulsar url explicitly and probably auth info also. ## Modifications Make the database.history.pulsar.service.url field required Add the config value back to example yaml files Update the integration test config

…he#11056) Fixes apache#8668 Expose `PulsarClient` via `BaseContext`, and allow connectors to use the inherited pulsar client from function worker to produce/consume messages. - [ ] Make sure that the change passes the CI checks. This change is already covered by existing tests, such as: - PulsarOffsetBackingStoreTest - KafkaConnectSourceTest - KafkaConnectSinkTest - The public API: `SourceContext` and `SinkContext` need to implement the `getPulsarClient` method (cherry picked from commit cb2ba71)

…he#11056) ### Motivation Fixes apache#8668 ### Modifications Expose `PulsarClient` via `BaseContext`, and allow connectors to use the inherited pulsar client from function worker to produce/consume messages. ### Verifying this change - [ ] Make sure that the change passes the CI checks. This change is already covered by existing tests, such as: - PulsarOffsetBackingStoreTest - KafkaConnectSourceTest - KafkaConnectSinkTest ### Does this pull request potentially affect one of the following parts: - The public API: `SourceContext` and `SinkContext` need to implement the `getPulsarClient` method

…y database (apache#11251) ### Motivation The Debezium requires pulsar a service URL for history database usage. In apache#11056 , the `service.url` field from `PulsarKafkaWorkerConfig` is no longer available. And the value is also deleted from multiple yaml config files in this [commit](apache@3ce24c9). This causes the integration test for Debezium connector to fail. Based on the Debezium [paradigm](https://debezium.io/documentation/reference/1.5/connectors/mysql.html#debezium-mysql-connector-database-history-configuration-properties), all configurations should be passed as strings. There's no easy way to inject a PulsarClient via configuration. We need to ask user to provide the pulsar url explicitly and probably auth info also. ### Modifications 1. Make the `database.history.pulsar.service.url` field required 2. Add the config value back to example yaml files 3. Update the integration test config ### Verifying this change - [ ] Make sure that the change passes the CI checks.

sijie assigned nlu90 Jun 24, 2021

sijie added the area/function label Jun 24, 2021

sijie added this to the 2.9.0 milestone Jun 24, 2021

sijie reviewed Jun 24, 2021

View reviewed changes

freeznet approved these changes Jun 24, 2021

View reviewed changes

eolivelli requested changes Jun 24, 2021

View reviewed changes

...a-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSinkTest.java Show resolved Hide resolved

dlg99 requested changes Jun 24, 2021

View reviewed changes

...nnect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarOffsetBackingStore.java Outdated Show resolved Hide resolved

pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/BaseContext.java Outdated Show resolved Hide resolved

nlu90 commented Jun 24, 2021

View reviewed changes

...a-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSinkTest.java Show resolved Hide resolved

...nnect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarOffsetBackingStore.java Outdated Show resolved Hide resolved

nlu90 marked this pull request as ready for review June 25, 2021 00:24

sijie approved these changes Jun 25, 2021

View reviewed changes

dlg99 approved these changes Jun 25, 2021

View reviewed changes

jerrypeng approved these changes Jun 25, 2021

View reviewed changes

nlu90 force-pushed the neng/io-pulsar-client branch from e6879f0 to 8991fdf Compare June 28, 2021 23:17

eolivelli approved these changes Jul 1, 2021

View reviewed changes

eolivelli changed the title ~~[pulsar-io] pass pulsar client via context to connector~~ PIP-85: [pulsar-io] pass pulsar client via context to connector Jul 1, 2021

nlu90 added 12 commits July 2, 2021 10:48

pass pulsar client via context to connector

f42b4f2

pass client via ctor

9663476

fix test

5b5aafb

update unit test

864133e

fix

ba39bd9

fix

16b738d

fix

35e4eb7

fix

449bc12

fix

bf99087

remove service url from kafka sink config

3ce24c9

fix unit test

28245b8

stop closing pulsar client in backing store

90b844b

nlu90 force-pushed the neng/io-pulsar-client branch from b894ebe to 90b844b Compare July 2, 2021 17:48

sijie merged commit 3939bce into apache:master Jul 2, 2021

sijie deleted the neng/io-pulsar-client branch July 2, 2021 18:53

dlg99 mentioned this pull request Jul 7, 2021

[Issue 8502] Upgrade Debezium to a newer version #11204

Merged

1 task

nlu90 mentioned this pull request Jul 7, 2021

[pulsar-io] pass the pulsar service url to debezium source for history database #11251

Merged

1 task

dlg99 mentioned this pull request Jul 15, 2021

Fixing Debezium integration tests #11154

Merged

1 task

Technoboy- mentioned this pull request Sep 21, 2021

[Cherry-pick] Pass client builder to debezium database history #12112

Merged

nicoloboschi mentioned this pull request Apr 5, 2022

[kca] Fix kca cherry-pick issue datastax/pulsar#63

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

PIP-85: [pulsar-io] pass pulsar client via context to connector #11056

PIP-85: [pulsar-io] pass pulsar client via context to connector #11056

nlu90 commented Jun 23, 2021 •

edited by sijie

Loading

sijie left a comment

eolivelli left a comment

dlg99 left a comment

dlg99 commented Jun 24, 2021

nlu90 left a comment

sijie commented Jun 25, 2021

dlg99 left a comment

jerrypeng commented Jun 25, 2021

dlg99 commented Jun 27, 2021

eolivelli commented Jun 28, 2021

eolivelli commented Jun 28, 2021

nlu90 commented Jun 28, 2021

eolivelli commented Jun 28, 2021

sijie commented Jun 28, 2021

eolivelli commented Jun 29, 2021

freeznet commented Jun 29, 2021

nlu90 commented Jun 30, 2021

sijie commented Jun 30, 2021

eolivelli commented Jul 1, 2021

eolivelli left a comment

nlu90 commented Jul 2, 2021

nlu90 commented Jul 2, 2021

PIP-85: [pulsar-io] pass pulsar client via context to connector #11056

PIP-85: [pulsar-io] pass pulsar client via context to connector #11056

Conversation

nlu90 commented Jun 23, 2021 • edited by sijie Loading

Motivation

Modifications

Verifying this change

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

sijie left a comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

dlg99 left a comment

Choose a reason for hiding this comment

dlg99 commented Jun 24, 2021

nlu90 left a comment

Choose a reason for hiding this comment

sijie commented Jun 25, 2021

dlg99 left a comment

Choose a reason for hiding this comment

jerrypeng commented Jun 25, 2021

dlg99 commented Jun 27, 2021

eolivelli commented Jun 28, 2021

eolivelli commented Jun 28, 2021

nlu90 commented Jun 28, 2021

eolivelli commented Jun 28, 2021

sijie commented Jun 28, 2021

eolivelli commented Jun 29, 2021

freeznet commented Jun 29, 2021

nlu90 commented Jun 30, 2021

sijie commented Jun 30, 2021

eolivelli commented Jul 1, 2021

eolivelli left a comment

Choose a reason for hiding this comment

nlu90 commented Jul 2, 2021

nlu90 commented Jul 2, 2021

nlu90 commented Jun 23, 2021 •

edited by sijie

Loading