You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/04/04 19:02:57 UTC
[GitHub] [pulsar] dlg99 opened a new pull request, #15025: [pulsar-io] KCA: handle whatever KeyValue getNativeObject() returns: org.apache.pulsar.io.core.KeyValue or org.apache.pulsar.common.schema.KeyValue
dlg99 opened a new pull request, #15025:
URL: https://github.com/apache/pulsar/pull/15025
### Motivation
KCA is confused between two KeyValue classes that we have: `org.apache.pulsar.common.schema.KeyValue` and `org.apache.pulsar.io.core.KeyValue`
`getNativeObject()` returns `common.schema.KeyValue` while the KCA expects `io.core.KeyValue`.
As result, sink fails with `ClassCastException: class org.apache.pulsar.common.schema.KeyValue cannot be cast to class org.apache.pulsar.io.core.KeyValue`
Long term it makes sense to get rid of the `io.core.KeyValue` (in the pulsar-io only) and replace its uses with newer `common.schema.KeyValue` (TBD)
### Modifications
For now I added support for both KeyValue types so the change can be cherry-picked into the older branches easily.
### Verifying this change
Added unit tests
### 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): (yes / no)
- The public API: (yes / no)
- The schema: (yes / no / don't know)
- The default values of configurations: (yes / no)
- The wire protocol: (yes / no)
- The rest endpoints: (yes / no)
- The admin cli options: (yes / no)
- Anything that affects deployment: (yes / no / don't know)
### Documentation
Check the box below or label this PR directly (if you have committer privilege).
Need to update docs?
- [ ] `doc-required`
(If you need help on updating docs, create a doc issue)
- [x] `no-need-doc`
(Please explain why)
- [ ] `doc`
(If this PR contains doc changes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] dlg99 commented on a diff in pull request #15025: [pulsar-io] KCA: properly handle KeyValue that getNativeObject() returns: corrected type + support for KeyValue
Posted by GitBox <gi...@apache.org>.
dlg99 commented on code in PR #15025:
URL: https://github.com/apache/pulsar/pull/15025#discussion_r844430107
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java:
##########
@@ -255,11 +254,25 @@ protected SinkRecord toSinkRecord(Record<GenericObject> sourceRecord) {
&& sourceRecord.getSchema().getSchemaInfo() != null
&& sourceRecord.getSchema().getSchemaInfo().getType() == SchemaType.KEY_VALUE) {
KeyValueSchema kvSchema = (KeyValueSchema) sourceRecord.getSchema();
- KeyValue kv = (KeyValue) sourceRecord.getValue().getNativeObject();
keySchema = PulsarSchemaToKafkaSchema.getKafkaConnectSchema(kvSchema.getKeySchema());
valueSchema = PulsarSchemaToKafkaSchema.getKafkaConnectSchema(kvSchema.getValueSchema());
- key = kv.getKey();
- value = kv.getValue();
+
+ Object nativeObject = sourceRecord.getValue().getNativeObject();
+
+ if (nativeObject instanceof org.apache.pulsar.common.schema.KeyValue) {
+ org.apache.pulsar.common.schema.KeyValue kv = (org.apache.pulsar.common.schema.KeyValue) nativeObject;
+ key = kv.getKey();
+ value = kv.getValue();
Review Comment:
@eolivelli fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] dlg99 commented on a diff in pull request #15025: [pulsar-io] KCA: handle whatever KeyValue getNativeObject() returns: org.apache.pulsar.io.core.KeyValue or org.apache.pulsar.common.schema.KeyValue
Posted by GitBox <gi...@apache.org>.
dlg99 commented on code in PR #15025:
URL: https://github.com/apache/pulsar/pull/15025#discussion_r842066211
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java:
##########
@@ -255,11 +254,25 @@ protected SinkRecord toSinkRecord(Record<GenericObject> sourceRecord) {
&& sourceRecord.getSchema().getSchemaInfo() != null
&& sourceRecord.getSchema().getSchemaInfo().getType() == SchemaType.KEY_VALUE) {
KeyValueSchema kvSchema = (KeyValueSchema) sourceRecord.getSchema();
- KeyValue kv = (KeyValue) sourceRecord.getValue().getNativeObject();
keySchema = PulsarSchemaToKafkaSchema.getKafkaConnectSchema(kvSchema.getKeySchema());
valueSchema = PulsarSchemaToKafkaSchema.getKafkaConnectSchema(kvSchema.getValueSchema());
- key = kv.getKey();
- value = kv.getValue();
+
+ Object nativeObject = sourceRecord.getValue().getNativeObject();
+
+ if (nativeObject instanceof org.apache.pulsar.common.schema.KeyValue) {
+ org.apache.pulsar.common.schema.KeyValue kv = (org.apache.pulsar.common.schema.KeyValue) nativeObject;
+ key = kv.getKey();
+ value = kv.getValue();
Review Comment:
I'd leave this anyway until we decide if it is ok to get rid of io.core.KeyValue altogether as a fix for https://github.com/apache/pulsar/issues/15027
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a diff in pull request #15025: [pulsar-io] KCA: handle whatever KeyValue getNativeObject() returns: org.apache.pulsar.io.core.KeyValue or org.apache.pulsar.common.schema.KeyValue
Posted by GitBox <gi...@apache.org>.
eolivelli commented on code in PR #15025:
URL: https://github.com/apache/pulsar/pull/15025#discussion_r842060050
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java:
##########
@@ -255,11 +254,25 @@ protected SinkRecord toSinkRecord(Record<GenericObject> sourceRecord) {
&& sourceRecord.getSchema().getSchemaInfo() != null
&& sourceRecord.getSchema().getSchemaInfo().getType() == SchemaType.KEY_VALUE) {
KeyValueSchema kvSchema = (KeyValueSchema) sourceRecord.getSchema();
- KeyValue kv = (KeyValue) sourceRecord.getValue().getNativeObject();
keySchema = PulsarSchemaToKafkaSchema.getKafkaConnectSchema(kvSchema.getKeySchema());
valueSchema = PulsarSchemaToKafkaSchema.getKafkaConnectSchema(kvSchema.getValueSchema());
- key = kv.getKey();
- value = kv.getValue();
+
+ Object nativeObject = sourceRecord.getValue().getNativeObject();
+
+ if (nativeObject instanceof org.apache.pulsar.common.schema.KeyValue) {
+ org.apache.pulsar.common.schema.KeyValue kv = (org.apache.pulsar.common.schema.KeyValue) nativeObject;
+ key = kv.getKey();
+ value = kv.getValue();
Review Comment:
This is non possible.
The wrapper for KeyValue is created in AutoConsumeSchema.
This class is like a utility class in Pulsar IO. Probably there is some mock test that creates the wrong object.
I believe we can safely drop this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] dlg99 commented on a diff in pull request #15025: [pulsar-io] KCA: handle whatever KeyValue getNativeObject() returns: org.apache.pulsar.io.core.KeyValue or org.apache.pulsar.common.schema.KeyValue
Posted by GitBox <gi...@apache.org>.
dlg99 commented on code in PR #15025:
URL: https://github.com/apache/pulsar/pull/15025#discussion_r842064474
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java:
##########
@@ -255,11 +254,25 @@ protected SinkRecord toSinkRecord(Record<GenericObject> sourceRecord) {
&& sourceRecord.getSchema().getSchemaInfo() != null
&& sourceRecord.getSchema().getSchemaInfo().getType() == SchemaType.KEY_VALUE) {
KeyValueSchema kvSchema = (KeyValueSchema) sourceRecord.getSchema();
- KeyValue kv = (KeyValue) sourceRecord.getValue().getNativeObject();
keySchema = PulsarSchemaToKafkaSchema.getKafkaConnectSchema(kvSchema.getKeySchema());
valueSchema = PulsarSchemaToKafkaSchema.getKafkaConnectSchema(kvSchema.getValueSchema());
- key = kv.getKey();
- value = kv.getValue();
+
+ Object nativeObject = sourceRecord.getValue().getNativeObject();
+
+ if (nativeObject instanceof org.apache.pulsar.common.schema.KeyValue) {
+ org.apache.pulsar.common.schema.KeyValue kv = (org.apache.pulsar.common.schema.KeyValue) nativeObject;
+ key = kv.getKey();
+ value = kv.getValue();
Review Comment:
@eolivelli did you mean that org.apache.pulsar.io.core.KeyValue is not possible?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli merged pull request #15025: [pulsar-io] KCA: properly handle KeyValue that getNativeObject() returns: corrected type + support for KeyValue
Posted by GitBox <gi...@apache.org>.
eolivelli merged PR #15025:
URL: https://github.com/apache/pulsar/pull/15025
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org