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/03/12 08:17:16 UTC
[GitHub] [pulsar] eolivelli opened a new pull request #9895: Implement Schema#OBJECT prototype
eolivelli opened a new pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895
Relates to #9844
### Motivation
Currently Schema#AUTO_CONSUME works only with GenericRecord, and makes it very difficult for users to deal with both Data Structures and primitive types because you have to know the type of data you are going to handle while creating the consumer or a Pulsar IO Sink.
This change introduces a Schema.OBJECT() (ObjectSchema.java) that is like Schema.AUTO_CONSUME but it is able to deal with GenericRecord, primitive types like Strings and KeyValue.
Currently KeyValue still does not work, due to another issue, that I am not going to address within this PR.
Currently this Patch does not change Pulsar IO, but once we agree on this way it will be straightforward to implement support for Sink<Object>
### Modifications
Add a new Schema.OBJECT() implementation, very like to Schema.AUTO_CONSUME()
### Verifying this change
Adding only minimal unit tests, I will add more tests once we agree on the direction.
----------------------------------------------------------------
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-804965045
@sijie @jerrypeng @merlimat @rdhabalia I hope that the GDoc is clarifying my usercases and why this patch is so important to me and to my users.
I hope we can move forward soon.
I will send out the Patch about PulsarSink after this preliminary 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 commented on pull request #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-801721286
@codelipenghui @sijie I have added support for PrimitiveRecord on top of this patch in this PR
#9956
----------------------------------------------------------------
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
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-797615746
@srkukarni can you please take a look ?
----------------------------------------------------------------
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-809937905
closed in favour of https://github.com/apache/pulsar/pull/10057
--
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-806563989
/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] sijie commented on pull request #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-801664166
The concerns will still be around AUTO_SCHEMA vs OBJECT. It will cause confusion. IMO, we should just have one schema type that is used to automatically get the "object" based on the schema type. You should think about `GenericRecord` as a wrapper over the "object" with schema information. A java `object` doesn't give you such schema type. Also "record" doesn't necessarily mean "struct", i.e In a log system a log record is a bag of bytes.
----------------------------------------------------------------
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-798923804
@codelipenghui
if you see this feature from a "new user"'s point of view, you will find this:
```
Consumer<Object> consumer = pulsarClient.newConsumer(Schema.OBJECT())
.subscriptionName("test")
.topic(topic)
.subscribe();
Message<Object> message = consumer.receive();
Object value = message.getValue();
```
that is really cool ! and it is self-explaining.
> why GenericRecord can't deal with the primary types?
Problems with GenericRecord approach:
1) we are adding a new wrapper object, and this is a waste of resources
2) A Record is a structure with "fields", but primitive types, lists and also KeyValue does not have "fields"
3) it is not intuitive that a String is a GenericRecord without "fields"
> In my opinion, the OBJECT schema will bring more complexity to the Pulsar Schema, 2 schema types want to deal with the same problem.
I believe that AUTO_CONSUME will still stay for the case that you know that you are dealing with Structures like AVRO, JSON, Protobuf, but if you want to write "generic" code, at least in Java, you really would like to use "Object" and not a custom Pulsar interface.
So for new users I imagine that their eyes will fall into "Object" as first chance, and they will discover "AUTO_CONSUME" only if they are interested in dealing directly with GenericRecord
In Avro itself you cannot map a primitive value or an array type to GenericRecord, see:
https://avro.apache.org/docs/current/api/java/org/apache/avro/generic/GenericRecord.html
If we really want a generic container I would add a new interface, like GenericObject o ObjectWithSchema
```
interface ObjectWithSchema {
Object asObject();
Schema getSchema();
}
```
By the way we cannot retrofit AUTO_CONSUME to Schema<ObjectWithSchema> as it is not a compatible change.
As we need to add a new "name" to the API (Schema.XXXX) let be this new name "OBJECT", that is very simple for Java users to understand
----------------------------------------------------------------
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
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-797471489
/pulsarbot rerun-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 closed pull request #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli closed pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895
--
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-798913224
> GenericRecord itself is a general purpose keyvalue dictionary, and using it to represent primitive types, arrays, KeyValue objects... is not intuitive and it will also require lots of plumbing and adapters
Yes, Looks a little weird. But with this approach, we don't need to introduce a new schema which is very similar to AUTO_CONSUME schema. This makes it difficult for users to decide whether to use AUTO_CONSUME schema or OBJECT schema. And use OBJECT schema also can achieve AUTO_CONSUME. From the perspective of schema API, we really need to consider it carefully.
The AUTO_CONSUME schema wants to deal with consuming any messages from the topic. The data of the topic can be the struct data and primary data. So, the AUTO_CONSUME schema should able to deal with the primary schema. The problem is we are returning the `GenericRecord` which looks more like better dealing with structured data, not primary data. So the essential problem is the definition of the `GenericRecord`, not the `AUTO_CONSUME` schema.
In my opinion, the OBJECT schema will bring more complexity to the Pulsar Schema, 2 schema types want to deal with the same problem. The `GenericRecord` maybe not a good name, If we have to make trade-offs, I prefer the GenericRecord approach. Take a step back, why `GenericRecord` can't deal with the primary types?
----------------------------------------------------------------
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-801656118
@sijie did you see the latest version of this patch here?
I have refactored the initial patch, now there is no code duplication and we have AbstractAutoConsumeSchema instead of AutoConsumeSchema.
In all of the points we checked for AutoConsumeSchema now we check for AbstractAutoConsumeSchema.
IMHO your concern has been addressed this way
----------------------------------------------------------------
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-803964281
@jerrypeng @merlimat this is the GDoc that explains the approach
https://docs.google.com/document/d/117gr6hG3N46MlkqENRKRtLejb_e_K0cZagQZv3jbwvQ/edit#
it also contains the PrimitiveRecord approach from @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] eolivelli commented on a change in pull request #9895: Implement Schema#OBJECT prototype
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#discussion_r593875446
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ObjectSchema.java
##########
@@ -0,0 +1,166 @@
+/**
+ * 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.client.impl.schema;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SchemaSerializationException;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
+import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+import java.util.concurrent.ExecutionException;
+
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Auto detect schema.
+ */
+@Slf4j
+public class ObjectSchema implements Schema<Object> {
+
+ private Schema<Object> schema;
+
+ private String topicName;
+
+ private String componentName;
+
+ private SchemaInfoProvider schemaInfoProvider;
+
+ public void setSchema(Schema<Object> schema) {
+ this.schema = schema;
+ }
+
+ private void ensureSchemaInitialized() {
+ checkState(null != schema, "Schema is not initialized before used");
+ }
+
+ @Override
+ public void validate(byte[] message) {
+ ensureSchemaInitialized();
+
+ schema.validate(message);
+ }
+
+ @Override
+ public boolean supportSchemaVersioning() {
+ return true;
+ }
+
+ @Override
+ public byte[] encode(Object message) {
+ ensureSchemaInitialized();
+
+ return schema.encode(message);
+ }
+
+ @Override
+ public Object decode(byte[] bytes, byte[] schemaVersion) {
+ if (schema == null) {
+ SchemaInfo schemaInfo = null;
+ try {
+ schemaInfo = schemaInfoProvider.getLatestSchema().get();
+ } catch (InterruptedException | ExecutionException e ) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ log.error("Con't get last schema for topic {} use AutoConsumeSchema", topicName);
+ throw new SchemaSerializationException(e.getCause());
+ }
+ schema = generateSchema(schemaInfo);
+ schema.setSchemaInfoProvider(schemaInfoProvider);
+ log.info("Configure {} schema for topic {} : {}",
+ componentName, topicName, schemaInfo.getSchemaDefinition());
+ }
+ ensureSchemaInitialized();
+ return schema.decode(bytes, schemaVersion);
+ }
+
+ @Override
+ public void setSchemaInfoProvider(SchemaInfoProvider schemaInfoProvider) {
+ if (schema == null) {
+ this.schemaInfoProvider = schemaInfoProvider;
+ } else {
+ schema.setSchemaInfoProvider(schemaInfoProvider);
+ }
+ }
+
+ @Override
+ public SchemaInfo getSchemaInfo() {
+ if (schema == null) {
+ return null;
+ }
+ return schema.getSchemaInfo();
+ }
+
+ @Override
+ public boolean requireFetchingSchemaInfo() {
+ return true;
+ }
+
+ @Override
+ public void configureSchemaInfo(String topicName,
+ String componentName,
+ SchemaInfo schemaInfo) {
+ this.topicName = topicName;
+ this.componentName = componentName;
+ if (schemaInfo != null) {
+ Schema<Object> genericSchema = generateSchema(schemaInfo);
+ setSchema(genericSchema);
+ log.info("Configure {} schema for topic {} : {}",
+ componentName, topicName, schemaInfo.getSchemaDefinition());
+ }
+ }
+
+ @Override
+ public Schema<Object> clone() {
+ Schema<Object> schema = Schema.OBJECT();
+ if (this.schema != null) {
+ schema.configureSchemaInfo(topicName, componentName, this.schema.getSchemaInfo());
+ } else {
+ schema.configureSchemaInfo(topicName, componentName, null);
+ }
+ if (schemaInfoProvider != null) {
+ schema.setSchemaInfoProvider(schemaInfoProvider);
+ }
+ return schema;
+ }
+
+ private Schema<Object> generateSchema(SchemaInfo schemaInfo) {
+ // when using `AutoConsumeSchema`, we use the schema associated with the messages as schema reader
+ // to decode the messages.
+ final boolean useProvidedSchemaAsReaderSchema = false;
+ switch (schemaInfo.getType()) {
+ case JSON:
+ case AVRO:
Review comment:
done
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ObjectSchema.java
##########
@@ -0,0 +1,166 @@
+/**
+ * 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.client.impl.schema;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SchemaSerializationException;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
+import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+import java.util.concurrent.ExecutionException;
+
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Auto detect schema.
+ */
+@Slf4j
+public class ObjectSchema implements Schema<Object> {
+
+ private Schema<Object> schema;
+
+ private String topicName;
+
+ private String componentName;
+
+ private SchemaInfoProvider schemaInfoProvider;
+
+ public void setSchema(Schema<Object> schema) {
+ this.schema = schema;
+ }
+
+ private void ensureSchemaInitialized() {
+ checkState(null != schema, "Schema is not initialized before used");
+ }
+
+ @Override
+ public void validate(byte[] message) {
+ ensureSchemaInitialized();
+
+ schema.validate(message);
+ }
+
+ @Override
+ public boolean supportSchemaVersioning() {
+ return true;
Review comment:
done
----------------------------------------------------------------
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-800877440
@sijie
I strongly believe that Object is the best choice for the long term:
1) Consumer<Object> and Sink<Object> is simpler for Java users
2) If we go down the wrapper way we should find another name, 'Record' is a type of data structure, IMHO a String does not fit the word "Record", the same for Array...(Avro itself does not wrap primitive types with GenericContainer, we can follow their model)
3) PrimitiveRecord wrapper won't match current behaviour of GenericRecord regarding nested structs (see below)
for 3): When you call "getField" on that GenericRecord for a String field now you receive a java.lang.String for a String and a GenericRecord for a Struct.
In this new model for a String field you should still return a PrimitiveRecord that wraps the String in order to be consistent, but it will be an important and unexpected behaviour change.
So all in all I believe that we should stick to Object type and not to a wrapper.
On Pulsar IO you already have the schema (Record#getSchema) and in the client API you can still lookup the schema using the Schema Version.
----------------------------------------------------------------
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 #9895: Implement Schema#OBJECT prototype
Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#discussion_r593493475
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ObjectSchema.java
##########
@@ -0,0 +1,166 @@
+/**
+ * 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.client.impl.schema;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SchemaSerializationException;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
+import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+import java.util.concurrent.ExecutionException;
+
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Auto detect schema.
+ */
+@Slf4j
+public class ObjectSchema implements Schema<Object> {
+
+ private Schema<Object> schema;
+
+ private String topicName;
+
+ private String componentName;
+
+ private SchemaInfoProvider schemaInfoProvider;
+
+ public void setSchema(Schema<Object> schema) {
+ this.schema = schema;
+ }
+
+ private void ensureSchemaInitialized() {
+ checkState(null != schema, "Schema is not initialized before used");
+ }
+
+ @Override
+ public void validate(byte[] message) {
+ ensureSchemaInitialized();
+
+ schema.validate(message);
+ }
+
+ @Override
+ public boolean supportSchemaVersioning() {
+ return true;
+ }
+
+ @Override
+ public byte[] encode(Object message) {
+ ensureSchemaInitialized();
+
+ return schema.encode(message);
+ }
+
+ @Override
+ public Object decode(byte[] bytes, byte[] schemaVersion) {
+ if (schema == null) {
+ SchemaInfo schemaInfo = null;
+ try {
+ schemaInfo = schemaInfoProvider.getLatestSchema().get();
+ } catch (InterruptedException | ExecutionException e ) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ log.error("Con't get last schema for topic {} use AutoConsumeSchema", topicName);
+ throw new SchemaSerializationException(e.getCause());
+ }
+ schema = generateSchema(schemaInfo);
+ schema.setSchemaInfoProvider(schemaInfoProvider);
+ log.info("Configure {} schema for topic {} : {}",
+ componentName, topicName, schemaInfo.getSchemaDefinition());
+ }
+ ensureSchemaInitialized();
+ return schema.decode(bytes, schemaVersion);
+ }
+
+ @Override
+ public void setSchemaInfoProvider(SchemaInfoProvider schemaInfoProvider) {
+ if (schema == null) {
+ this.schemaInfoProvider = schemaInfoProvider;
+ } else {
+ schema.setSchemaInfoProvider(schemaInfoProvider);
+ }
+ }
+
+ @Override
+ public SchemaInfo getSchemaInfo() {
+ if (schema == null) {
+ return null;
+ }
+ return schema.getSchemaInfo();
+ }
+
+ @Override
+ public boolean requireFetchingSchemaInfo() {
+ return true;
+ }
+
+ @Override
+ public void configureSchemaInfo(String topicName,
+ String componentName,
+ SchemaInfo schemaInfo) {
+ this.topicName = topicName;
+ this.componentName = componentName;
+ if (schemaInfo != null) {
+ Schema<Object> genericSchema = generateSchema(schemaInfo);
+ setSchema(genericSchema);
+ log.info("Configure {} schema for topic {} : {}",
+ componentName, topicName, schemaInfo.getSchemaDefinition());
+ }
+ }
+
+ @Override
+ public Schema<Object> clone() {
+ Schema<Object> schema = Schema.OBJECT();
+ if (this.schema != null) {
+ schema.configureSchemaInfo(topicName, componentName, this.schema.getSchemaInfo());
+ } else {
+ schema.configureSchemaInfo(topicName, componentName, null);
+ }
+ if (schemaInfoProvider != null) {
+ schema.setSchemaInfoProvider(schemaInfoProvider);
+ }
+ return schema;
+ }
+
+ private Schema<Object> generateSchema(SchemaInfo schemaInfo) {
+ // when using `AutoConsumeSchema`, we use the schema associated with the messages as schema reader
+ // to decode the messages.
+ final boolean useProvidedSchemaAsReaderSchema = false;
+ switch (schemaInfo.getType()) {
+ case JSON:
+ case AVRO:
Review comment:
there is also a PROTOBUF schema, no?
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ObjectSchema.java
##########
@@ -0,0 +1,166 @@
+/**
+ * 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.client.impl.schema;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SchemaSerializationException;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
+import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+import java.util.concurrent.ExecutionException;
+
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Auto detect schema.
+ */
+@Slf4j
+public class ObjectSchema implements Schema<Object> {
+
+ private Schema<Object> schema;
+
+ private String topicName;
+
+ private String componentName;
+
+ private SchemaInfoProvider schemaInfoProvider;
+
+ public void setSchema(Schema<Object> schema) {
+ this.schema = schema;
+ }
+
+ private void ensureSchemaInitialized() {
+ checkState(null != schema, "Schema is not initialized before used");
+ }
+
+ @Override
+ public void validate(byte[] message) {
+ ensureSchemaInitialized();
+
+ schema.validate(message);
+ }
+
+ @Override
+ public boolean supportSchemaVersioning() {
+ return true;
+ }
+
+ @Override
+ public byte[] encode(Object message) {
+ ensureSchemaInitialized();
+
+ return schema.encode(message);
+ }
+
+ @Override
+ public Object decode(byte[] bytes, byte[] schemaVersion) {
Review comment:
there is a lot of duplicated code from AutoConsumeSchema.
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ObjectSchema.java
##########
@@ -0,0 +1,166 @@
+/**
+ * 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.client.impl.schema;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SchemaSerializationException;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
+import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+import java.util.concurrent.ExecutionException;
+
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Auto detect schema.
+ */
+@Slf4j
+public class ObjectSchema implements Schema<Object> {
+
+ private Schema<Object> schema;
+
+ private String topicName;
+
+ private String componentName;
+
+ private SchemaInfoProvider schemaInfoProvider;
+
+ public void setSchema(Schema<Object> schema) {
+ this.schema = schema;
+ }
+
+ private void ensureSchemaInitialized() {
+ checkState(null != schema, "Schema is not initialized before used");
+ }
+
+ @Override
+ public void validate(byte[] message) {
+ ensureSchemaInitialized();
+
+ schema.validate(message);
+ }
+
+ @Override
+ public boolean supportSchemaVersioning() {
+ return true;
Review comment:
This is not entirely true here. For primitive schemas, this would be `false`.
----------------------------------------------------------------
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
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-798840367
@codelipenghui
You are suggesting to use GenericRecord as an interface that represents and kind of data structure.
Indeed I thought a lot about your same idea! But in the end I believe it does not fit well.
GenericRecord itself is a general purpose keyvalue dictionary, and using it to represent primitive types, arrays, KeyValue objects... is not intuitive and it will also require lots of plumbing and adapters
I believe that having a Schema that returns Object is really want you expect when you are dealing with generic code.
----------------------------------------------------------------
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
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-797426343
@codelipenghui @sijie @rdhabalia @merlimat
what do you think about this approach ?
if it makes sense I will add more tests and complete the 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] jerrypeng commented on pull request #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-802587065
Having a Schema of Object type does not make sense to me. The point of a schema is to provide boundaries on what the structure of the data can be. However, having a schema that supports a generic POJO that can have any structure seems paradoxical to me. What is the difference between this and just not having a schema at all?
--
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
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-797770946
@merlimat if this patch is on the right way I will move forward with adding tests.
I would also to create a prototype about running Pulsar IO `Sink<Object>` based on this work an on #9896
----------------------------------------------------------------
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-800411617
/pulsarbot rerun-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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-802613355
@jerrypeng the point here is about having a way to build **general purpose** Consumers and especially Sinks (I will send a follow up patch for Pulsar IO once we decide the right way for this feature) without knowing at compile time the schema.
So this is not about not using a Schema, but the point is to be able to consume all of the Schemas automatically.
If I do not set a schema, I can only see a byte[], and I lose all of the information about the type of data.
We already have Schema.AUTO_CONSUME() and downloads automatically the schema and decodes the payload to a higher level model.
This patch is only about extending AUTO_CONSUME() in order to deal with primitive types and with KeyValue (and ProtobufNative, but this is another story).
Unfortunately we cannot change the signature of AUTO_CONSUME from Schema<GenericRecord> to Schema<Object>, so I am adding a new static method Schema.OBJECT().
I am putting up a Google Doc in order to illustrate all of the aspects of the problem I want to solve and which are the alternatives
--
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
codelipenghui commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-798823545
We added `Object getNativeRecord()` in the GenericRecord, is it better to reuse this method to support primitive types? Looks like
```java
if (record.getSchemaType() == SchemaType.STRING) {
String value = (String) record.getNativeRecord();
}
```
----------------------------------------------------------------
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-798929750
/pulsarbot rerun-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] sijie commented on pull request #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-801606979
`getField` will return an empty list. `schemaType` gives the schema type. `GenericRecord` is just a container to hold either a primitive value or a struct. My main point here is to avoid introducing duplicated code between #AUTO_SCHEMA and #OBJECT. They share a lot of similarities. I want to add duplicated checking logic overall the place.
----------------------------------------------------------------
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 #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-800815707
Why not return a PrimitiveRecord that implements a `GenericRecord` interface?
`getSchemaType` will give you the right schema type and `getNativeRecord` will give you the right underlying object. So a lot of logic that already deal with AUTO_SCHEMA would almost just work. We don't need to worry about there is another `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] eolivelli commented on pull request #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-800278355
/pulsarbot rerun-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] jerrypeng commented on pull request #9895: Implement Schema#OBJECT - support for Consumer
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#issuecomment-803019382
The problem seems to be with `getSchema()` in Record and the subsequent `getSchema()` on the actual pulsar message. For consumer declared like Consumer<byte[]>, getSchema() calls on messages the consumer returns will always be SCHEMA.BYTES, regardless of what the actual schema the message was produced with. The schema of message is overridden when byte[] is specified as the schema for the consumer. This seems like a bug to me. Perhaps @sijie can also chime in on why the behavior is such.
If messages returned by a Consumer<byte[]> had the right schema, i.e. the schema of the topic / schema the message was produced with, then your problem is solved right @eolivelli ?
--
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 #9895: Implement Schema#OBJECT prototype
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9895:
URL: https://github.com/apache/pulsar/pull/9895#discussion_r593719356
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ObjectSchema.java
##########
@@ -0,0 +1,166 @@
+/**
+ * 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.client.impl.schema;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SchemaSerializationException;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
+import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+import java.util.concurrent.ExecutionException;
+
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Auto detect schema.
+ */
+@Slf4j
+public class ObjectSchema implements Schema<Object> {
+
+ private Schema<Object> schema;
+
+ private String topicName;
+
+ private String componentName;
+
+ private SchemaInfoProvider schemaInfoProvider;
+
+ public void setSchema(Schema<Object> schema) {
+ this.schema = schema;
+ }
+
+ private void ensureSchemaInitialized() {
+ checkState(null != schema, "Schema is not initialized before used");
+ }
+
+ @Override
+ public void validate(byte[] message) {
+ ensureSchemaInitialized();
+
+ schema.validate(message);
+ }
+
+ @Override
+ public boolean supportSchemaVersioning() {
+ return true;
+ }
+
+ @Override
+ public byte[] encode(Object message) {
+ ensureSchemaInitialized();
+
+ return schema.encode(message);
+ }
+
+ @Override
+ public Object decode(byte[] bytes, byte[] schemaVersion) {
Review comment:
@sijie thank you for taking a look to this prototype so early.
From your comments I understand that this work is interesting for you and we are on the same page.
This patch is only a prototype, my idea is to create a common class for AutoConsumeSchema and ObjectSchema,
now I only copy/pasted from AutoConsumeSchema in order to just show to the community what I am doing.
Unfortunately we cannot switch AutoConsumeSchema from `Schema<GenericRecord>` to `Schema<Object>` because it will be an incompatible change (at least at the source level) and this will cause lot of troubles to users that are expecting only to receive GenericRecord instances from AUTO_CONSUME.
I will finish the work and I will ping you when I am done
----------------------------------------------------------------
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