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 2021/02/04 13:44:30 UTC
[GitHub] [pulsar] eolivelli opened a new pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
eolivelli opened a new pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481
### Motivation
Currently Pulsar Source Connectors cannot produce messages using the GenericRecord API.
Producing messages with the GenericRecord API will allow to dynamically generate data structures that can be consumed downstream using the supported encoding
### Modifications
Allow a Pulsar Source to be declared to produce "GenericRecord" instances, this in turn means to change PulsarSink that is the entity that receives message from the Source and writes to the Pulsar topic.
- PulsarSink: do not precreate a Schema in case of GenericRecord data type (the schema will be created at the first write)
- AvroWriter: support encoding any GenericRecord and not only GenericAvroRecords
- Add integration test
### Verifying this change
The change adds integration tests and new unit tests
### Documentation
There is no need to add documentation, because it comes naturally for a developer to declare the source as `Source<GenericRecord>` and it is expected that it works
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572148075
##########
File path: pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
##########
@@ -196,6 +197,9 @@ private static boolean isProtobufClass(Class<?> pojoClazz) {
String schemaTypeOrClassName = conf.getSchemaType();
if (StringUtils.isEmpty(schemaTypeOrClassName) || DEFAULT_SERDE.equals(schemaTypeOrClassName)) {
// No preferred schema was provided, auto-discover schema or fallback to defaults
+ if (!input && clazz.equals(GenericRecord.class)) {
+ return new AutoProduceBytesSchema();
+ }
Review comment:
It seems integration test don't cover here, it still use `record.getSchema()`. What is the integration test testing? I did not understand here.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r571002075
##########
File path: pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/TopicSchemaTest.java
##########
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.functions.source;
+
+import lombok.Cleanup;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.*;
Review comment:
oh sure.
I will update the patch after @congbobo184 review, in order to save CI cycles
thanks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572670613
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroWriter.java
##########
@@ -52,4 +67,29 @@ public GenericAvroWriter(Schema schema) {
this.byteArrayOutputStream.reset();
}
}
+
+ /**
+ * This is an adapter from Pulsar GenericRecord to Avro classes.
+ */
+ private class GenericRecordAdapter extends SpecificRecordBase {
+ private GenericRecord message;
+
+ void setCurrentMessage(GenericRecord message) {
+ this.message = message;
+ }
+ @Override
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public Object get(int field) {
+ return message.getField(schema.getFields().get(field).name());
+ }
+
+ @Override
+ public void put(int field, Object value) {
+ throw new UnsupportedOperationException();
+ }
+ }
Review comment:
It also can change the message in GenericRecordAdapter, GenericRecordAdapter seems to have no effect. It can't prevent change.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-776693960
thank you @congbobo184
can you please commit this patch ?
this way I can rebase another patch of mine and send it
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572670613
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroWriter.java
##########
@@ -52,4 +67,29 @@ public GenericAvroWriter(Schema schema) {
this.byteArrayOutputStream.reset();
}
}
+
+ /**
+ * This is an adapter from Pulsar GenericRecord to Avro classes.
+ */
+ private class GenericRecordAdapter extends SpecificRecordBase {
+ private GenericRecord message;
+
+ void setCurrentMessage(GenericRecord message) {
+ this.message = message;
+ }
+ @Override
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public Object get(int field) {
+ return message.getField(schema.getFields().get(field).name());
+ }
+
+ @Override
+ public void put(int field, Object value) {
+ throw new UnsupportedOperationException();
+ }
+ }
Review comment:
It also can change the message in GenericRecordAdapter, GenericRecordAdapter seems to have no effect
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572242740
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroWriter.java
##########
@@ -52,4 +67,29 @@ public GenericAvroWriter(Schema schema) {
this.byteArrayOutputStream.reset();
}
}
+
+ /**
+ * This is an adapter from Pulsar GenericRecord to Avro classes.
+ */
+ private class GenericRecordAdapter extends SpecificRecordBase {
+ private GenericRecord message;
+
+ void setCurrentMessage(GenericRecord message) {
+ this.message = message;
+ }
+ @Override
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public Object get(int field) {
+ return message.getField(schema.getFields().get(field).name());
+ }
+
+ @Override
+ public void put(int field, Object value) {
+ throw new UnsupportedOperationException();
+ }
+ }
Review comment:
Here we are in a writer and a writer is supposed to only read from the object and not to alter the contents.
So it is good here that we prevent any change.
I cannot understand your point.
The integration tests makes an example of a Souce that is pushing GenericRecord instances and then those messages can be consumed by a standard Pulsar consumer with AUTO_CONSUME
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572718212
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroWriter.java
##########
@@ -52,4 +67,29 @@ public GenericAvroWriter(Schema schema) {
this.byteArrayOutputStream.reset();
}
}
+
+ /**
+ * This is an adapter from Pulsar GenericRecord to Avro classes.
+ */
+ private class GenericRecordAdapter extends SpecificRecordBase {
+ private GenericRecord message;
+
+ void setCurrentMessage(GenericRecord message) {
+ this.message = message;
+ }
+ @Override
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public Object get(int field) {
+ return message.getField(schema.getFields().get(field).name());
+ }
+
+ @Override
+ public void put(int field, Object value) {
+ throw new UnsupportedOperationException();
+ }
+ }
Review comment:
I understand what you mean, but `GenericDatumWriter` can write `GenericRecord`. Should we make GenericRecordAdapter implement `GenericRecord` not `SpecificRecordBase`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-780015838
@codelipenghui: Most of the understandings are correct. There is one mistake. We don't use `AUTO_CONSUME` for serializing the messages. We just use `AUTO_CONSUME` to tell PulsarSink that it doesn't need to pre-create a producer. The messages produced to the sink are all `GenericRecord`s. They all carry the schema information themselves.
Then we leverage PIP-45 to write the messages based on the schema information they carry.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-777262457
@sijie @codelipenghui
@congbobo184 approved this patch.
can we merge it please ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-776072651
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-775941560
/pulsarbot run-failure-tests
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] wolfstudy commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
wolfstudy commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572664845
##########
File path: tests/docker-images/java-test-functions/src/main/java/org/apache/pulsar/tests/integration/io/TestGenericRecordSource.java
##########
@@ -0,0 +1,102 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.io;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.*;
Review comment:
@eolivelli Please fix 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-779885238
@sijie Thanks for the comment. If I understand correctly, the approach is using the AUTO_CONSUME schema and the PIP-43 to achieve this purpose. Previously, we always use the AUTO_CONSUME schema for a consumer. Essentially, the AUTO_CONSUME schema is a Generic Schema that works with GenericRecord, encode a GenericRecord to byte[] and decode byte[] to GenericRecord. So we don't need to change the existing implementation, we can use the AUTO_CONSUME schema for a producer too to allow the producer able to publish GenericRecord.
And with PIP-45, the message carries the real schema. So that we are able to publish GenericRecord with different version schemas.
Please point out if I am wrong, thanks.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-776686497
LGTM! great work!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli closed pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli closed pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r570723440
##########
File path: pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
##########
@@ -385,28 +388,43 @@ public void close() throws Exception {
}
}
+ @AllArgsConstructor
+ static class InitSchemaResult<T> {
+ final Schema<T> schema;
+ final boolean requireSink;
+ }
+
@SuppressWarnings("unchecked")
@VisibleForTesting
- Schema<T> initializeSchema() throws ClassNotFoundException {
+ InitSchemaResult<T> initializeSchema() throws ClassNotFoundException {
if (StringUtils.isEmpty(this.pulsarSinkConfig.getTypeClassName())) {
- return (Schema<T>) Schema.BYTES;
+ return new InitSchemaResult((Schema<T>) Schema.BYTES, true);
}
Class<?> typeArg = Reflections.loadClass(this.pulsarSinkConfig.getTypeClassName(), functionClassLoader);
if (Void.class.equals(typeArg)) {
// return type is 'void', so there's no schema to check
- return null;
+ log.info("typeClassName is {}, no need to force a schema", this.pulsarSinkConfig.getTypeClassName());
+ return new InitSchemaResult(null, false);
+ }
+ if (GenericRecord.class.equals(typeArg)) {
Review comment:
I am not sure this implementation here is correct. The schema should be inferred by `TopicSchema.getSchema`. You should be able to obtain the right schema instance by inferring it from the `typeArg`, `sinkSchemaType` (`sinkSchemaType` is same as the `sourceSchemaType`).
I think the right direction is to fix https://github.com/apache/pulsar/blob/0469dfe2c7804bd9ca9ea34e95d83b2196216cf9/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java#L146, make sure we construct the right schema when type is `GenericRecord`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572817733
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroWriter.java
##########
@@ -52,4 +67,29 @@ public GenericAvroWriter(Schema schema) {
this.byteArrayOutputStream.reset();
}
}
+
+ /**
+ * This is an adapter from Pulsar GenericRecord to Avro classes.
+ */
+ private class GenericRecordAdapter extends SpecificRecordBase {
+ private GenericRecord message;
+
+ void setCurrentMessage(GenericRecord message) {
+ this.message = message;
+ }
+ @Override
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public Object get(int field) {
+ return message.getField(schema.getFields().get(field).name());
+ }
+
+ @Override
+ public void put(int field, Object value) {
+ throw new UnsupportedOperationException();
+ }
+ }
Review comment:
@congbobo184 I have double checked, and we indeed only need to implement GenericRecord.
The usage of SpecificRecordBase was a left-over of my initial implementation.
I have simplified the code by implementing all of the GenericRecord fields.
good catch !
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572676915
##########
File path: pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
##########
@@ -196,6 +197,9 @@ private static boolean isProtobufClass(Class<?> pojoClazz) {
String schemaTypeOrClassName = conf.getSchemaType();
if (StringUtils.isEmpty(schemaTypeOrClassName) || DEFAULT_SERDE.equals(schemaTypeOrClassName)) {
// No preferred schema was provided, auto-discover schema or fallback to defaults
+ if (!input && clazz.equals(GenericRecord.class)) {
+ return new AutoProduceBytesSchema();
+ }
Review comment:
@congbobo184
This method is called by `PulsarSink` in `initializeSchema() `method via `TopicSchema#getSchema`
The PulsarSink is the entity that actually write to Pulsar.
When you create a Source Pulsar IO creates a PulsarSink to write to the destination topic.
When you start the Source you are going to use this special `AutoProduceBytesSchema` (it was pre-existing, I did not add it).
Initially the Source does not enforce a Schema on the topic (we achieve this with `AutoProduceBytesSchema`).
When the Source passes a Record to the Pulsar runtime, the PulsarSink picks up the Schema (using Record#getSchema) and sets the schema properly.
Therefore when the schema changes (in a compatible way) the runtime automatically updates the schema.
In fact when the Source starts you cannot know the schema, because the schema is generated dynamically by the Source itself.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-779514341
For the reviewers on this pull request, please take a look at #9590. In order to support writing a generic record, we don't need to change the existing AVRO schema implementation. It should be done in PulsarSink, as it needs to support writing multiple versions of generic records. Pulsar Sink already has the framework to support that by leveraging [PIP-43](https://github.com/apache/pulsar/wiki/PIP-43%3A-producer-send-message-with-different-schema).
#9590 is ready for review with unit tests. integration tests to be added.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-776503858
@congbobo184 CI passed, I hope that this patch is good to go now
thank you again for your review @congbobo184 @codelipenghui and @sijie
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572722501
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroWriter.java
##########
@@ -52,4 +67,29 @@ public GenericAvroWriter(Schema schema) {
this.byteArrayOutputStream.reset();
}
}
+
+ /**
+ * This is an adapter from Pulsar GenericRecord to Avro classes.
+ */
+ private class GenericRecordAdapter extends SpecificRecordBase {
+ private GenericRecord message;
+
+ void setCurrentMessage(GenericRecord message) {
+ this.message = message;
+ }
+ @Override
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public Object get(int field) {
+ return message.getField(schema.getFields().get(field).name());
+ }
+
+ @Override
+ public void put(int field, Object value) {
+ throw new UnsupportedOperationException();
+ }
+ }
Review comment:
why we make `GenericRecordAdapter` extend `SpecificRecordBase` not `GenericRecord `, seem `private final GenericDatumWriter<org.apache.avro.generic.GenericRecord> writer;` only can write `GenericRecord` not `SpecificRecordBase`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572725567
##########
File path: pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
##########
@@ -196,6 +197,9 @@ private static boolean isProtobufClass(Class<?> pojoClazz) {
String schemaTypeOrClassName = conf.getSchemaType();
if (StringUtils.isEmpty(schemaTypeOrClassName) || DEFAULT_SERDE.equals(schemaTypeOrClassName)) {
// No preferred schema was provided, auto-discover schema or fallback to defaults
+ if (!input && clazz.equals(GenericRecord.class)) {
+ return new AutoProduceBytesSchema();
+ }
Review comment:
Thank you for your detailed reply, I understand here. :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572241040
##########
File path: pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
##########
@@ -196,6 +197,9 @@ private static boolean isProtobufClass(Class<?> pojoClazz) {
String schemaTypeOrClassName = conf.getSchemaType();
if (StringUtils.isEmpty(schemaTypeOrClassName) || DEFAULT_SERDE.equals(schemaTypeOrClassName)) {
// No preferred schema was provided, auto-discover schema or fallback to defaults
+ if (!input && clazz.equals(GenericRecord.class)) {
+ return new AutoProduceBytesSchema();
+ }
Review comment:
The integration test without this change does not work.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli edited a comment on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli edited a comment on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-775941560
/pulsarbot run-failures-tests
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r571000579
##########
File path: pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/TopicSchemaTest.java
##########
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.functions.source;
+
+import lombok.Cleanup;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.*;
Review comment:
It's better to avoid star import.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-774931137
@congbobo184 do you have cycles ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-774935260
@codelipenghui I have addressed your comment and merged with current master.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572682722
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroWriter.java
##########
@@ -52,4 +67,29 @@ public GenericAvroWriter(Schema schema) {
this.byteArrayOutputStream.reset();
}
}
+
+ /**
+ * This is an adapter from Pulsar GenericRecord to Avro classes.
+ */
+ private class GenericRecordAdapter extends SpecificRecordBase {
+ private GenericRecord message;
+
+ void setCurrentMessage(GenericRecord message) {
+ this.message = message;
+ }
+ @Override
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public Object get(int field) {
+ return message.getField(schema.getFields().get(field).name());
+ }
+
+ @Override
+ public void put(int field, Object value) {
+ throw new UnsupportedOperationException();
+ }
+ }
Review comment:
here we are adding an adapter from a Pulsar GenericRecord to SpecificRecordBase but we are interested in implementing only read-only methods.
if you prefer I can leave the implementation of this method empty, the method won't be called by the writer.
If you run the integration test you will see that we are passing thru this code.
Current implementation in master assumes that the GenericRecord is always a GenericAvroRecord, but in case of this new feature it can be any implementation of GenericRecord (and in fact in my case it will be a generic data structure generated dynamically, with a dynamic schema)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-775964246
/pulsarbot run-failure-tests
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-774058398
@congbobo184 Could you please also help review this PR?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572143808
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroWriter.java
##########
@@ -52,4 +67,29 @@ public GenericAvroWriter(Schema schema) {
this.byteArrayOutputStream.reset();
}
}
+
+ /**
+ * This is an adapter from Pulsar GenericRecord to Avro classes.
+ */
+ private class GenericRecordAdapter extends SpecificRecordBase {
+ private GenericRecord message;
+
+ void setCurrentMessage(GenericRecord message) {
+ this.message = message;
+ }
+ @Override
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public Object get(int field) {
+ return message.getField(schema.getFields().get(field).name());
+ }
+
+ @Override
+ public void put(int field, Object value) {
+ throw new UnsupportedOperationException();
+ }
+ }
Review comment:
Now we generate writer is `GenericDatumWriter`, it can't write `SpecificRecordBase` , we should support SpecificDatumWriter. So, I don't think it a good way to do this.
##########
File path: pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
##########
@@ -196,6 +197,9 @@ private static boolean isProtobufClass(Class<?> pojoClazz) {
String schemaTypeOrClassName = conf.getSchemaType();
if (StringUtils.isEmpty(schemaTypeOrClassName) || DEFAULT_SERDE.equals(schemaTypeOrClassName)) {
// No preferred schema was provided, auto-discover schema or fallback to defaults
+ if (!input && clazz.equals(GenericRecord.class)) {
+ return new AutoProduceBytesSchema();
+ }
Review comment:
It seems integration test don't cover here, it also use `record.getSchema()`. What is the integration test testing? I did not understand here.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572718212
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroWriter.java
##########
@@ -52,4 +67,29 @@ public GenericAvroWriter(Schema schema) {
this.byteArrayOutputStream.reset();
}
}
+
+ /**
+ * This is an adapter from Pulsar GenericRecord to Avro classes.
+ */
+ private class GenericRecordAdapter extends SpecificRecordBase {
+ private GenericRecord message;
+
+ void setCurrentMessage(GenericRecord message) {
+ this.message = message;
+ }
+ @Override
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public Object get(int field) {
+ return message.getField(schema.getFields().get(field).name());
+ }
+
+ @Override
+ public void put(int field, Object value) {
+ throw new UnsupportedOperationException();
+ }
+ }
Review comment:
I understand what you mean, but `GenericDatumWriter` can write `GenericRecord`. Should we make `GenericRecordAdapter` implement `GenericRecord` not `SpecificRecordBase`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572242740
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroWriter.java
##########
@@ -52,4 +67,29 @@ public GenericAvroWriter(Schema schema) {
this.byteArrayOutputStream.reset();
}
}
+
+ /**
+ * This is an adapter from Pulsar GenericRecord to Avro classes.
+ */
+ private class GenericRecordAdapter extends SpecificRecordBase {
+ private GenericRecord message;
+
+ void setCurrentMessage(GenericRecord message) {
+ this.message = message;
+ }
+ @Override
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public Object get(int field) {
+ return message.getField(schema.getFields().get(field).name());
+ }
+
+ @Override
+ public void put(int field, Object value) {
+ throw new UnsupportedOperationException();
+ }
+ }
Review comment:
Here we are in a writer and a writer is supposed to only read from the object and not to alter the contents.
So it is good here that we prevent any change.
I cannot understand your point.
The integration tests makes an example of a Source that is pushing GenericRecord instances and then those messages can be consumed by a standard Pulsar consumer with AUTO_CONSUME
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-779497237
@eolivelli I think the change can be as simple as https://github.com/apache/pulsar/pull/9590. I haven't verified it works. But it should be the right direction to go. Because PulsarSink already handles writing messages using multiple schemas (https://github.com/apache/pulsar/blob/c27f7a340a3c78a3a5004d08b8ac05ed99672ee3/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java#L230). For GenericRecord source connectors, each source record already carries the schema information. The PulsarSink can use that schema information to write a the record.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] rdhabalia commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
rdhabalia commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-779035465
@sijie can you please help to review this PR.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r570977387
##########
File path: pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
##########
@@ -385,28 +388,43 @@ public void close() throws Exception {
}
}
+ @AllArgsConstructor
+ static class InitSchemaResult<T> {
+ final Schema<T> schema;
+ final boolean requireSink;
+ }
+
@SuppressWarnings("unchecked")
@VisibleForTesting
- Schema<T> initializeSchema() throws ClassNotFoundException {
+ InitSchemaResult<T> initializeSchema() throws ClassNotFoundException {
if (StringUtils.isEmpty(this.pulsarSinkConfig.getTypeClassName())) {
- return (Schema<T>) Schema.BYTES;
+ return new InitSchemaResult((Schema<T>) Schema.BYTES, true);
}
Class<?> typeArg = Reflections.loadClass(this.pulsarSinkConfig.getTypeClassName(), functionClassLoader);
if (Void.class.equals(typeArg)) {
// return type is 'void', so there's no schema to check
- return null;
+ log.info("typeClassName is {}, no need to force a schema", this.pulsarSinkConfig.getTypeClassName());
+ return new InitSchemaResult(null, false);
+ }
+ if (GenericRecord.class.equals(typeArg)) {
Review comment:
@sijie thank you for your suggestion.
I have updated the patch according this comment.
I have also tested manually and it works like a charm.
The integration test asserts that everything is working properly
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r570723440
##########
File path: pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
##########
@@ -385,28 +388,43 @@ public void close() throws Exception {
}
}
+ @AllArgsConstructor
+ static class InitSchemaResult<T> {
+ final Schema<T> schema;
+ final boolean requireSink;
+ }
+
@SuppressWarnings("unchecked")
@VisibleForTesting
- Schema<T> initializeSchema() throws ClassNotFoundException {
+ InitSchemaResult<T> initializeSchema() throws ClassNotFoundException {
if (StringUtils.isEmpty(this.pulsarSinkConfig.getTypeClassName())) {
- return (Schema<T>) Schema.BYTES;
+ return new InitSchemaResult((Schema<T>) Schema.BYTES, true);
}
Class<?> typeArg = Reflections.loadClass(this.pulsarSinkConfig.getTypeClassName(), functionClassLoader);
if (Void.class.equals(typeArg)) {
// return type is 'void', so there's no schema to check
- return null;
+ log.info("typeClassName is {}, no need to force a schema", this.pulsarSinkConfig.getTypeClassName());
+ return new InitSchemaResult(null, false);
+ }
+ if (GenericRecord.class.equals(typeArg)) {
Review comment:
I am not sure this implementation here is correct. The schema should be inferred by `TopicSchema.getSchema`. You should be able to obtain the right schema instance by inferring it from the `typeArg`, `sinkSchemaType` (`sinkSchemaType` is same as the `sourceSchemaType`).
I think the right direction is to fix https://github.com/apache/pulsar/blob/0469dfe2c7804bd9ca9ea34e95d83b2196216cf9/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java#L146, make sure we construct the right schema when type is `GenericRecord`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572678310
##########
File path: tests/docker-images/java-test-functions/src/main/java/org/apache/pulsar/tests/integration/io/TestGenericRecordSource.java
##########
@@ -0,0 +1,102 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.io;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.*;
Review comment:
sure, done @wolfstudy
thanks for catching 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#discussion_r572669296
##########
File path: pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
##########
@@ -196,6 +197,9 @@ private static boolean isProtobufClass(Class<?> pojoClazz) {
String schemaTypeOrClassName = conf.getSchemaType();
if (StringUtils.isEmpty(schemaTypeOrClassName) || DEFAULT_SERDE.equals(schemaTypeOrClassName)) {
// No preferred schema was provided, auto-discover schema or fallback to defaults
+ if (!input && clazz.equals(GenericRecord.class)) {
+ return new AutoProduceBytesSchema();
+ }
Review comment:
you start a source, but source don't getSchema by this method. I don't understand where use this change?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-775879696
@congbobo184 I have addressed your comment PTAL, thanks for your review
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #9481: Allow Pulsar IO Sources to push GenericRecord instances encoded with AVRO
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9481:
URL: https://github.com/apache/pulsar/pull/9481#issuecomment-780366916
I am closing this PR in favour of #9590.
I would like to see the integration tests in this patch added to #9590 as they show an usecase I have to support
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org