You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@pulsar.apache.org by Christophe Bornet <bo...@gmail.com> on 2023/03/27 17:15:04 UTC
Re: [DISCUSS] forbid user to upload `BYTES` schema
This change broke the Flink SQL Pulsar connector:
https://github.com/streamnative/flink/issues/270
So I propose to revert it.
Le ven. 9 déc. 2022 à 11:57, labuladong <la...@foxmail.com> a écrit :
>
> Hi pulsar community,
>
>
> I'd like to discuss the behavior of schema uploading, for more context see https://github.com/apache/pulsar/issues/18825
>
>
> I think that forbidding users to upload `BYTES` schema is a recommended way to solve this issue. But this may change the existing behavior, so do you have any suggestion about this issue?
>
>
> Thanks,
> Donglai
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by PengHui Li <pe...@apache.org>.
I guess the initial purpose was to reduce the RPC calls and
the workload of the schema registry (each operation will perform
at Znode update). And I agree with saving the BYTES schema is
more clear than saving noting.
For now, the challenge is the client side will skip uploading the BYTES
schema, not the broker side rejecting the BYTES schema uploading.
And we are not able to force requires the client to upload the BYTES schema,
because it will break the old client versions. This can be a very long
process
to ensure all the topics will not have an empty schema.
Regards,
Penghui
On Thu, Jun 1, 2023 at 8:02 AM Neng Lu <fr...@gmail.com> wrote:
> Treating the BYTES schema differently (i.e. forbid uploading it) than other
> schemas will be confusing to users.
>
> If the BYTES schema is the default schema for a new topic, then how about
> saving a BYTES schema in the registry instead of saving nothing?
>
>
>
> On Mon, Apr 17, 2023 at 8:42 PM PengHui Li <pe...@apache.org> wrote:
>
> > I have pushed out a PR to revert this change first.
> >
> > https://github.com/apache/pulsar/pull/20123
> >
> > Please help review.
> >
> > Thanks,
> > Penghui
> >
> > On Tue, Apr 18, 2023 at 11:36 AM PengHui Li <pe...@apache.org> wrote:
> >
> > > > Flink uses the schema to store some kv based properties. If we can
> > > expose all the operations of the topic level metadata store. We can
> > > truly drop the use of uploading the BYTES schema in the Flink
> > > connector.
> > >
> > > After 3.0.0, Pulsar will provide the ability to set properties for a
> > topic
> > > https://github.com/apache/pulsar/pull/17238.
> > >
> > > Thanks,
> > > Penghui
> > >
> > > On Mon, Apr 17, 2023 at 11:44 PM Yufan Sheng <sy...@gmail.com> wrote:
> > >
> > >> >I support reverting the PR first and then looking for a long-term
> > >> solution.
> > >>
> > >> Flink uses the schema to store some kv based properties. If we can
> > >> expose all the operations of the topic level metadata store. We can
> > >> truly drop the use of uploading the BYTES schema in the Flink
> > >> connector.
> > >>
> > >> On Mon, Apr 17, 2023 at 12:12 PM PengHui Li <pe...@apache.org>
> wrote:
> > >> >
> > >> > I'm sorry. I have provided the wrong description of the changes from
> > >> the PR.
> > >> > The PR has changed the server side, so it's hard for users to ask to
> > >> > upgrade
> > >> > the Flink connector if the pulsar server is upgraded.
> > >> >
> > >> > I support reverting the PR first and then looking for a long-term
> > >> solution.
> > >> >
> > >> > Best,
> > >> > Penghui
> > >> >
> > >> > On Mon, Apr 17, 2023 at 10:16 AM PengHui Li <pe...@apache.org>
> > wrote:
> > >> >
> > >> > > > Did we consider
> > >> > > making a call to upload a Bytes schema a no-op?
> > >> > >
> > >> > > It was a BUG that the PR fixed.
> > >> > > You will not be able to get the uploaded schema as expected.
> > >> > > Please take a look at the details from the GitHub issue.
> > >> > >
> > >> > > What is the challenge for the Flink connector now?
> > >> > > The changes only take effect on the client side.
> > >> > > So, the issue will only happen if they use a new connector.
> > >> > > Upgrading the Pulsar server will not make any impaction?
> > >> > > Is it better to fix the Flink connector?
> > >> > > IMO, the Flink connector should not use admin-api
> > >> > > to upload a BYTE schema. It's a redundant operation.
> > >> > > Pulsar will do nothing.
> > >> > >
> > >> > > What do you think about a long-term solution?
> > >> > >
> > >> > > Regards
> > >> > > - Penghui
> > >> > >
> > >> > > On Sat, Apr 15, 2023 at 12:52 AM Michael Marshall <
> > >> mmarshall@apache.org>
> > >> > > wrote:
> > >> > >
> > >> > >> I think the primary point is that unless there is a strict need,
> we
> > >> > >> shouldn't introduce breaking changes to the implementation. Why
> did
> > >> we
> > >> > >> choose to forbid users from uploading a Bytes schema? Did we
> > consider
> > >> > >> making a call to upload a Bytes schema a no-op?
> > >> > >>
> > >> > >> Thanks,
> > >> > >> Michael
> > >> > >>
> > >> > >> On Fri, Apr 14, 2023 at 10:46 AM SiNan Liu <
> liusinan1998@gmail.com
> > >
> > >> > >> wrote:
> > >> > >> >
> > >> > >> > 1. I don't know much about flink, but what I see here is that
> you
> > >> need
> > >> > >> to
> > >> > >> > save a `ResolvedCatalogTable`, which I see has `CatalogTable`,
> so
> > >> it is
> > >> > >> > used to record the metadata information of the table.
> > >> > >> > **In
> > >> > >> >
> > >> > >>
> > >>
> >
> org.apache.flink.connector.pulsar.table.catalog.impl.PulsarCatalogSupport#createTable**
> > >> > >> > ```java
> > >> > >> > @Override
> > >> > >> > public void createTable(ObjectPath tablePath,
> > >> ResolvedCatalogTable
> > >> > >> > table)
> > >> > >> > throws PulsarAdminException {
> > >> > >> > // only allow creating table in explict database, the
> > >> topic is
> > >> > >> used
> > >> > >> > to save table
> > >> > >> > // information
> > >> > >> > if (!isExplicitDatabase(tablePath.getDatabaseName())) {
> > >> > >> > throw new CatalogException(
> > >> > >> > String.format(
> > >> > >> > "Can't create explict table under
> > >> pulsar
> > >> > >> > tenant/namespace: %s because it's a native database",
> > >> > >> > tablePath.getDatabaseName()));
> > >> > >> > }
> > >> > >> >
> > >> > >> > String mappedTopic =
> > >> > >> findExplicitTablePlaceholderTopic(tablePath);
> > >> > >> > pulsarAdminTool.createTopic(mappedTopic, 1);
> > >> > >> >
> > >> > >> > // use pulsar schema to store explicit table
> information
> > >> > >> > try {
> > >> > >> > SchemaInfo schemaInfo =
> > >> > >> > TableSchemaHelper.generateSchemaInfo(table.toProperties());
> > >> > >> > pulsarAdminTool.uploadSchema(mappedTopic,
> > schemaInfo);
> > >> > >> > } catch (Exception e) {
> > >> > >> > // delete topic if table info cannot be persisted
> > >> > >> > try {
> > >> > >> > pulsarAdminTool.deleteTopic(mappedTopic);
> > >> > >> > } catch (PulsarAdminException ex) {
> > >> > >> > // do nothing
> > >> > >> > }
> > >> > >> > e.printStackTrace();
> > >> > >> > throw new CatalogException("Can't store table
> > >> metadata");
> > >> > >> > }
> > >> > >> > }
> > >> > >> > ```
> > >> > >> >
> > >> > >> > 2. In
> > >> `TableSchemaHelper.generateSchemaInfo(table.toProperties());`：
> > >> > >> > Why must SchemaType.BYTE be used? Is it OK to use
> > SchemaType.JSON?
> > >> > >> > ```java
> > >> > >> > public static SchemaInfo generateSchemaInfo(Map<String,
> > String>
> > >> > >> > properties)
> > >> > >> > throws JsonProcessingException {
> > >> > >> > ObjectMapper mapper = new ObjectMapper();
> > >> > >> > // json
> > >> > >> > String json = mapper.writeValueAsString(properties);
> > >> > >> > return SchemaInfoImpl.builder()
> > >> > >> > .name("flink_table_schema")
> > >> > >> > //.type(SchemaType.BYTES)
> > >> > >> > //.schema(mapper.writeValueAsBytes(properties))
> > >> > >> > // SchemaType.JSON
> > >> > >> > .type(SchemaType.JSON)
> > >> > >> > .schema(json.getBytes())
> > >> > >> > .build();
> > >> > >> > }
> > >> > >> > ```
> > >> > >> >
> > >> > >> > 3. Sorry, I'm a newbie. Can an experienced developer help with
> > >> this?
> > >> > >> Thanks!
> > >> > >> >
> > >> > >> >
> > >> > >> > Thanks,
> > >> > >> > sinan
> > >> > >> >
> > >> > >> >
> > >> > >> > Enrico Olivelli <eo...@gmail.com> 于2023年4月14日周五 22:50写道：
> > >> > >> >
> > >> > >> > > Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet
> > >> > >> > > <bo...@gmail.com> ha scritto:
> > >> > >> > > >
> > >> > >> > > > The change was not reverted for 3.0.0-RC so in the current
> > >> state the
> > >> > >> > > > Flink Pulsar catalog won't work with Pulsar 3.0.
> > >> > >> > > > To give more details, Flink Pulsar stores Flink Table
> > metadata
> > >> in
> > >> > >> the
> > >> > >> > > > metadata of BYTES schema topics (those topics don't have
> > >> messages,
> > >> > >> > > > only the schema metadata of these topics is used).
> > >> > >> > >
> > >> > >> > > I think that we should not introduce this breaking change, it
> > >> will be
> > >> > >> > > a big pain for the users in the Flink ecosystem
> > >> > >> > >
> > >> > >> > > Enrico
> > >> > >> > >
> > >> > >> > > >
> > >> > >> > > > Le mer. 29 mars 2023 à 09:54, Yufan Sheng <
> syhily@gmail.com>
> > a
> > >> > >> écrit :
> > >> > >> > > > >
> > >> > >> > > > > Hi SiNan,
> > >> > >> > > > >
> > >> > >> > > > > In the flink world, we don't always rely on the schema
> > >> information
> > >> > >> > > > > provided by Pulsar or other connector systems. Flink
> > >> application
> > >> > >> has
> > >> > >> > > > > its own (de)serialization schema logic, which treats the
> > >> messages
> > >> > >> only
> > >> > >> > > > > in a binary format like a byte array.
> > >> > >> > > > >
> > >> > >> > > > > In flink-connector-pulsar, we only use the schema when
> the
> > >> users
> > >> > >> want
> > >> > >> > > > > to do some evolution check. Otherwise, we will only send
> > >> messages
> > >> > >> in
> > >> > >> > > > > BYTES schema.
> > >> > >> > > > >
> > >> > >> > > > > On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <
> > >> > >> liusinan1998@gmail.com>
> > >> > >> > > wrote:
> > >> > >> > > > > >
> > >> > >> > > > > > Hi yufan.
> > >> > >> > > > > > Can you describe a bit the usage scenario of byte
> schema
> > in
> > >> > >> > > > > > flink-connector-pulsa?
> > >> > >> > > > > >
> > >> > >> > > > > >
> > >> > >> > > > > > Thanks,
> > >> > >> > > > > > sinan
> > >> > >> > > > > >
> > >> > >> > > > > > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二
> 上午9:53写道：
> > >> > >> > > > > >
> > >> > >> > > > > > > As the flink-connector-pulsar developer, I don't want
> > to
> > >> > >> disable
> > >> > >> > > the
> > >> > >> > > > > > > BYTES schema upload. In my opinion, using BYTES
> schema
> > >> means
> > >> > >> the
> > >> > >> > > users
> > >> > >> > > > > > > want to bypass the schema check and handle the schema
> > >> > >> validation by
> > >> > >> > > > > > > themselves.
> > >> > >> > > > > > >
> > >> > >> > > > > > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <
> > >> > >> liusinan1998@gmail.com>
> > >> > >> > > wrote:
> > >> > >> > > > > > > >
> > >> > >> > > > > > > > Hi, everyone.
> > >> > >> > > > > > > > When a user uploads bytes schema. We can warn the
> > user
> > >> and
> > >> > >> skip
> > >> > >> > > uploading
> > >> > >> > > > > > > > bytes schema.
> > >> > >> > > > > > > > Also check to see if the topic has a schema other
> > than
> > >> > >> bytes.
> > >> > >> > > > > > > > 1. If yes, warn the user that it is not necessary
> to
> > >> upload
> > >> > >> > > bytes schema.
> > >> > >> > > > > > > > You can subscribe to a topic using bytes schema.
> > >> > >> > > > > > > > 2. If there is no schema, warn the user that the
> > topic
> > >> does
> > >> > >> not
> > >> > >> > > have a
> > >> > >> > > > > > > > schema. The default is bytes schema, and there is
> no
> > >> need to
> > >> > >> > > upload it.
> > >> > >> > > > > > > > Rather than simply throwing an exception rejecting
> > the
> > >> > >> upload
> > >> > >> > > bytes
> > >> > >> > > > > > > schema.
> > >> > >> > > > > > > >
> > >> > >> > > > > > > >
> > >> > >> > > > > > > > Thanks,
> > >> > >> > > > > > > > sinan
> > >> > >> > > > > > > >
> > >> > >> > > > > > > >
> > >> > >> > > > > > > > Christophe Bornet <bo...@gmail.com> 于
> > >> 2023年3月28日周二
> > >> > >> > > 上午1:15写道：
> > >> > >> > > > > > > >
> > >> > >> > > > > > > > > This change broke the Flink SQL Pulsar connector:
> > >> > >> > > > > > > > > https://github.com/streamnative/flink/issues/270
> > >> > >> > > > > > > > > So I propose to revert it.
> > >> > >> > > > > > > > >
> > >> > >> > > > > > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <
> > >> > >> > > labuladong@foxmail.com> a
> > >> > >> > > > > > > écrit :
> > >> > >> > > > > > > > > >
> > >> > >> > > > > > > > > > Hi pulsar community,
> > >> > >> > > > > > > > > >
> > >> > >> > > > > > > > > >
> > >> > >> > > > > > > > > > I'd like to discuss the behavior of schema
> > >> uploading,
> > >> > >> for
> > >> > >> > > more
> > >> > >> > > > > > > context
> > >> > >> > > > > > > > > see
> https://github.com/apache/pulsar/issues/18825
> > >> > >> > > > > > > > > >
> > >> > >> > > > > > > > > >
> > >> > >> > > > > > > > > > I think that forbidding users to upload `BYTES`
> > >> schema
> > >> > >> is a
> > >> > >> > > > > > > recommended
> > >> > >> > > > > > > > > way to solve this issue. But this may change the
> > >> existing
> > >> > >> > > behavior, so
> > >> > >> > > > > > > do
> > >> > >> > > > > > > > > you have any suggestion about this issue?
> > >> > >> > > > > > > > > >
> > >> > >> > > > > > > > > >
> > >> > >> > > > > > > > > > Thanks,
> > >> > >> > > > > > > > > > Donglai
> > >> > >> > > > > > > > >
> > >> > >> > > > > > >
> > >> > >> > >
> > >> > >>
> > >> > >
> > >>
> > >
> >
>
>
> --
> Best Regards,
> Neng
>
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by Neng Lu <fr...@gmail.com>.
Treating the BYTES schema differently (i.e. forbid uploading it) than other
schemas will be confusing to users.
If the BYTES schema is the default schema for a new topic, then how about
saving a BYTES schema in the registry instead of saving nothing?
On Mon, Apr 17, 2023 at 8:42 PM PengHui Li <pe...@apache.org> wrote:
> I have pushed out a PR to revert this change first.
>
> https://github.com/apache/pulsar/pull/20123
>
> Please help review.
>
> Thanks,
> Penghui
>
> On Tue, Apr 18, 2023 at 11:36 AM PengHui Li <pe...@apache.org> wrote:
>
> > > Flink uses the schema to store some kv based properties. If we can
> > expose all the operations of the topic level metadata store. We can
> > truly drop the use of uploading the BYTES schema in the Flink
> > connector.
> >
> > After 3.0.0, Pulsar will provide the ability to set properties for a
> topic
> > https://github.com/apache/pulsar/pull/17238.
> >
> > Thanks,
> > Penghui
> >
> > On Mon, Apr 17, 2023 at 11:44 PM Yufan Sheng <sy...@gmail.com> wrote:
> >
> >> >I support reverting the PR first and then looking for a long-term
> >> solution.
> >>
> >> Flink uses the schema to store some kv based properties. If we can
> >> expose all the operations of the topic level metadata store. We can
> >> truly drop the use of uploading the BYTES schema in the Flink
> >> connector.
> >>
> >> On Mon, Apr 17, 2023 at 12:12 PM PengHui Li <pe...@apache.org> wrote:
> >> >
> >> > I'm sorry. I have provided the wrong description of the changes from
> >> the PR.
> >> > The PR has changed the server side, so it's hard for users to ask to
> >> > upgrade
> >> > the Flink connector if the pulsar server is upgraded.
> >> >
> >> > I support reverting the PR first and then looking for a long-term
> >> solution.
> >> >
> >> > Best,
> >> > Penghui
> >> >
> >> > On Mon, Apr 17, 2023 at 10:16 AM PengHui Li <pe...@apache.org>
> wrote:
> >> >
> >> > > > Did we consider
> >> > > making a call to upload a Bytes schema a no-op?
> >> > >
> >> > > It was a BUG that the PR fixed.
> >> > > You will not be able to get the uploaded schema as expected.
> >> > > Please take a look at the details from the GitHub issue.
> >> > >
> >> > > What is the challenge for the Flink connector now?
> >> > > The changes only take effect on the client side.
> >> > > So, the issue will only happen if they use a new connector.
> >> > > Upgrading the Pulsar server will not make any impaction?
> >> > > Is it better to fix the Flink connector?
> >> > > IMO, the Flink connector should not use admin-api
> >> > > to upload a BYTE schema. It's a redundant operation.
> >> > > Pulsar will do nothing.
> >> > >
> >> > > What do you think about a long-term solution?
> >> > >
> >> > > Regards
> >> > > - Penghui
> >> > >
> >> > > On Sat, Apr 15, 2023 at 12:52 AM Michael Marshall <
> >> mmarshall@apache.org>
> >> > > wrote:
> >> > >
> >> > >> I think the primary point is that unless there is a strict need, we
> >> > >> shouldn't introduce breaking changes to the implementation. Why did
> >> we
> >> > >> choose to forbid users from uploading a Bytes schema? Did we
> consider
> >> > >> making a call to upload a Bytes schema a no-op?
> >> > >>
> >> > >> Thanks,
> >> > >> Michael
> >> > >>
> >> > >> On Fri, Apr 14, 2023 at 10:46 AM SiNan Liu <liusinan1998@gmail.com
> >
> >> > >> wrote:
> >> > >> >
> >> > >> > 1. I don't know much about flink, but what I see here is that you
> >> need
> >> > >> to
> >> > >> > save a `ResolvedCatalogTable`, which I see has `CatalogTable`, so
> >> it is
> >> > >> > used to record the metadata information of the table.
> >> > >> > **In
> >> > >> >
> >> > >>
> >>
> org.apache.flink.connector.pulsar.table.catalog.impl.PulsarCatalogSupport#createTable**
> >> > >> > ```java
> >> > >> > @Override
> >> > >> > public void createTable(ObjectPath tablePath,
> >> ResolvedCatalogTable
> >> > >> > table)
> >> > >> > throws PulsarAdminException {
> >> > >> > // only allow creating table in explict database, the
> >> topic is
> >> > >> used
> >> > >> > to save table
> >> > >> > // information
> >> > >> > if (!isExplicitDatabase(tablePath.getDatabaseName())) {
> >> > >> > throw new CatalogException(
> >> > >> > String.format(
> >> > >> > "Can't create explict table under
> >> pulsar
> >> > >> > tenant/namespace: %s because it's a native database",
> >> > >> > tablePath.getDatabaseName()));
> >> > >> > }
> >> > >> >
> >> > >> > String mappedTopic =
> >> > >> findExplicitTablePlaceholderTopic(tablePath);
> >> > >> > pulsarAdminTool.createTopic(mappedTopic, 1);
> >> > >> >
> >> > >> > // use pulsar schema to store explicit table information
> >> > >> > try {
> >> > >> > SchemaInfo schemaInfo =
> >> > >> > TableSchemaHelper.generateSchemaInfo(table.toProperties());
> >> > >> > pulsarAdminTool.uploadSchema(mappedTopic,
> schemaInfo);
> >> > >> > } catch (Exception e) {
> >> > >> > // delete topic if table info cannot be persisted
> >> > >> > try {
> >> > >> > pulsarAdminTool.deleteTopic(mappedTopic);
> >> > >> > } catch (PulsarAdminException ex) {
> >> > >> > // do nothing
> >> > >> > }
> >> > >> > e.printStackTrace();
> >> > >> > throw new CatalogException("Can't store table
> >> metadata");
> >> > >> > }
> >> > >> > }
> >> > >> > ```
> >> > >> >
> >> > >> > 2. In
> >> `TableSchemaHelper.generateSchemaInfo(table.toProperties());`：
> >> > >> > Why must SchemaType.BYTE be used? Is it OK to use
> SchemaType.JSON?
> >> > >> > ```java
> >> > >> > public static SchemaInfo generateSchemaInfo(Map<String,
> String>
> >> > >> > properties)
> >> > >> > throws JsonProcessingException {
> >> > >> > ObjectMapper mapper = new ObjectMapper();
> >> > >> > // json
> >> > >> > String json = mapper.writeValueAsString(properties);
> >> > >> > return SchemaInfoImpl.builder()
> >> > >> > .name("flink_table_schema")
> >> > >> > //.type(SchemaType.BYTES)
> >> > >> > //.schema(mapper.writeValueAsBytes(properties))
> >> > >> > // SchemaType.JSON
> >> > >> > .type(SchemaType.JSON)
> >> > >> > .schema(json.getBytes())
> >> > >> > .build();
> >> > >> > }
> >> > >> > ```
> >> > >> >
> >> > >> > 3. Sorry, I'm a newbie. Can an experienced developer help with
> >> this?
> >> > >> Thanks!
> >> > >> >
> >> > >> >
> >> > >> > Thanks,
> >> > >> > sinan
> >> > >> >
> >> > >> >
> >> > >> > Enrico Olivelli <eo...@gmail.com> 于2023年4月14日周五 22:50写道：
> >> > >> >
> >> > >> > > Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet
> >> > >> > > <bo...@gmail.com> ha scritto:
> >> > >> > > >
> >> > >> > > > The change was not reverted for 3.0.0-RC so in the current
> >> state the
> >> > >> > > > Flink Pulsar catalog won't work with Pulsar 3.0.
> >> > >> > > > To give more details, Flink Pulsar stores Flink Table
> metadata
> >> in
> >> > >> the
> >> > >> > > > metadata of BYTES schema topics (those topics don't have
> >> messages,
> >> > >> > > > only the schema metadata of these topics is used).
> >> > >> > >
> >> > >> > > I think that we should not introduce this breaking change, it
> >> will be
> >> > >> > > a big pain for the users in the Flink ecosystem
> >> > >> > >
> >> > >> > > Enrico
> >> > >> > >
> >> > >> > > >
> >> > >> > > > Le mer. 29 mars 2023 à 09:54, Yufan Sheng <sy...@gmail.com>
> a
> >> > >> écrit :
> >> > >> > > > >
> >> > >> > > > > Hi SiNan,
> >> > >> > > > >
> >> > >> > > > > In the flink world, we don't always rely on the schema
> >> information
> >> > >> > > > > provided by Pulsar or other connector systems. Flink
> >> application
> >> > >> has
> >> > >> > > > > its own (de)serialization schema logic, which treats the
> >> messages
> >> > >> only
> >> > >> > > > > in a binary format like a byte array.
> >> > >> > > > >
> >> > >> > > > > In flink-connector-pulsar, we only use the schema when the
> >> users
> >> > >> want
> >> > >> > > > > to do some evolution check. Otherwise, we will only send
> >> messages
> >> > >> in
> >> > >> > > > > BYTES schema.
> >> > >> > > > >
> >> > >> > > > > On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <
> >> > >> liusinan1998@gmail.com>
> >> > >> > > wrote:
> >> > >> > > > > >
> >> > >> > > > > > Hi yufan.
> >> > >> > > > > > Can you describe a bit the usage scenario of byte schema
> in
> >> > >> > > > > > flink-connector-pulsa?
> >> > >> > > > > >
> >> > >> > > > > >
> >> > >> > > > > > Thanks,
> >> > >> > > > > > sinan
> >> > >> > > > > >
> >> > >> > > > > > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
> >> > >> > > > > >
> >> > >> > > > > > > As the flink-connector-pulsar developer, I don't want
> to
> >> > >> disable
> >> > >> > > the
> >> > >> > > > > > > BYTES schema upload. In my opinion, using BYTES schema
> >> means
> >> > >> the
> >> > >> > > users
> >> > >> > > > > > > want to bypass the schema check and handle the schema
> >> > >> validation by
> >> > >> > > > > > > themselves.
> >> > >> > > > > > >
> >> > >> > > > > > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <
> >> > >> liusinan1998@gmail.com>
> >> > >> > > wrote:
> >> > >> > > > > > > >
> >> > >> > > > > > > > Hi, everyone.
> >> > >> > > > > > > > When a user uploads bytes schema. We can warn the
> user
> >> and
> >> > >> skip
> >> > >> > > uploading
> >> > >> > > > > > > > bytes schema.
> >> > >> > > > > > > > Also check to see if the topic has a schema other
> than
> >> > >> bytes.
> >> > >> > > > > > > > 1. If yes, warn the user that it is not necessary to
> >> upload
> >> > >> > > bytes schema.
> >> > >> > > > > > > > You can subscribe to a topic using bytes schema.
> >> > >> > > > > > > > 2. If there is no schema, warn the user that the
> topic
> >> does
> >> > >> not
> >> > >> > > have a
> >> > >> > > > > > > > schema. The default is bytes schema, and there is no
> >> need to
> >> > >> > > upload it.
> >> > >> > > > > > > > Rather than simply throwing an exception rejecting
> the
> >> > >> upload
> >> > >> > > bytes
> >> > >> > > > > > > schema.
> >> > >> > > > > > > >
> >> > >> > > > > > > >
> >> > >> > > > > > > > Thanks,
> >> > >> > > > > > > > sinan
> >> > >> > > > > > > >
> >> > >> > > > > > > >
> >> > >> > > > > > > > Christophe Bornet <bo...@gmail.com> 于
> >> 2023年3月28日周二
> >> > >> > > 上午1:15写道：
> >> > >> > > > > > > >
> >> > >> > > > > > > > > This change broke the Flink SQL Pulsar connector:
> >> > >> > > > > > > > > https://github.com/streamnative/flink/issues/270
> >> > >> > > > > > > > > So I propose to revert it.
> >> > >> > > > > > > > >
> >> > >> > > > > > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <
> >> > >> > > labuladong@foxmail.com> a
> >> > >> > > > > > > écrit :
> >> > >> > > > > > > > > >
> >> > >> > > > > > > > > > Hi pulsar community,
> >> > >> > > > > > > > > >
> >> > >> > > > > > > > > >
> >> > >> > > > > > > > > > I'd like to discuss the behavior of schema
> >> uploading,
> >> > >> for
> >> > >> > > more
> >> > >> > > > > > > context
> >> > >> > > > > > > > > see https://github.com/apache/pulsar/issues/18825
> >> > >> > > > > > > > > >
> >> > >> > > > > > > > > >
> >> > >> > > > > > > > > > I think that forbidding users to upload `BYTES`
> >> schema
> >> > >> is a
> >> > >> > > > > > > recommended
> >> > >> > > > > > > > > way to solve this issue. But this may change the
> >> existing
> >> > >> > > behavior, so
> >> > >> > > > > > > do
> >> > >> > > > > > > > > you have any suggestion about this issue?
> >> > >> > > > > > > > > >
> >> > >> > > > > > > > > >
> >> > >> > > > > > > > > > Thanks,
> >> > >> > > > > > > > > > Donglai
> >> > >> > > > > > > > >
> >> > >> > > > > > >
> >> > >> > >
> >> > >>
> >> > >
> >>
> >
>
--
Best Regards,
Neng
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by PengHui Li <pe...@apache.org>.
I have pushed out a PR to revert this change first.
https://github.com/apache/pulsar/pull/20123
Please help review.
Thanks,
Penghui
On Tue, Apr 18, 2023 at 11:36 AM PengHui Li <pe...@apache.org> wrote:
> > Flink uses the schema to store some kv based properties. If we can
> expose all the operations of the topic level metadata store. We can
> truly drop the use of uploading the BYTES schema in the Flink
> connector.
>
> After 3.0.0, Pulsar will provide the ability to set properties for a topic
> https://github.com/apache/pulsar/pull/17238.
>
> Thanks,
> Penghui
>
> On Mon, Apr 17, 2023 at 11:44 PM Yufan Sheng <sy...@gmail.com> wrote:
>
>> >I support reverting the PR first and then looking for a long-term
>> solution.
>>
>> Flink uses the schema to store some kv based properties. If we can
>> expose all the operations of the topic level metadata store. We can
>> truly drop the use of uploading the BYTES schema in the Flink
>> connector.
>>
>> On Mon, Apr 17, 2023 at 12:12 PM PengHui Li <pe...@apache.org> wrote:
>> >
>> > I'm sorry. I have provided the wrong description of the changes from
>> the PR.
>> > The PR has changed the server side, so it's hard for users to ask to
>> > upgrade
>> > the Flink connector if the pulsar server is upgraded.
>> >
>> > I support reverting the PR first and then looking for a long-term
>> solution.
>> >
>> > Best,
>> > Penghui
>> >
>> > On Mon, Apr 17, 2023 at 10:16 AM PengHui Li <pe...@apache.org> wrote:
>> >
>> > > > Did we consider
>> > > making a call to upload a Bytes schema a no-op?
>> > >
>> > > It was a BUG that the PR fixed.
>> > > You will not be able to get the uploaded schema as expected.
>> > > Please take a look at the details from the GitHub issue.
>> > >
>> > > What is the challenge for the Flink connector now?
>> > > The changes only take effect on the client side.
>> > > So, the issue will only happen if they use a new connector.
>> > > Upgrading the Pulsar server will not make any impaction?
>> > > Is it better to fix the Flink connector?
>> > > IMO, the Flink connector should not use admin-api
>> > > to upload a BYTE schema. It's a redundant operation.
>> > > Pulsar will do nothing.
>> > >
>> > > What do you think about a long-term solution?
>> > >
>> > > Regards
>> > > - Penghui
>> > >
>> > > On Sat, Apr 15, 2023 at 12:52 AM Michael Marshall <
>> mmarshall@apache.org>
>> > > wrote:
>> > >
>> > >> I think the primary point is that unless there is a strict need, we
>> > >> shouldn't introduce breaking changes to the implementation. Why did
>> we
>> > >> choose to forbid users from uploading a Bytes schema? Did we consider
>> > >> making a call to upload a Bytes schema a no-op?
>> > >>
>> > >> Thanks,
>> > >> Michael
>> > >>
>> > >> On Fri, Apr 14, 2023 at 10:46 AM SiNan Liu <li...@gmail.com>
>> > >> wrote:
>> > >> >
>> > >> > 1. I don't know much about flink, but what I see here is that you
>> need
>> > >> to
>> > >> > save a `ResolvedCatalogTable`, which I see has `CatalogTable`, so
>> it is
>> > >> > used to record the metadata information of the table.
>> > >> > **In
>> > >> >
>> > >>
>> org.apache.flink.connector.pulsar.table.catalog.impl.PulsarCatalogSupport#createTable**
>> > >> > ```java
>> > >> > @Override
>> > >> > public void createTable(ObjectPath tablePath,
>> ResolvedCatalogTable
>> > >> > table)
>> > >> > throws PulsarAdminException {
>> > >> > // only allow creating table in explict database, the
>> topic is
>> > >> used
>> > >> > to save table
>> > >> > // information
>> > >> > if (!isExplicitDatabase(tablePath.getDatabaseName())) {
>> > >> > throw new CatalogException(
>> > >> > String.format(
>> > >> > "Can't create explict table under
>> pulsar
>> > >> > tenant/namespace: %s because it's a native database",
>> > >> > tablePath.getDatabaseName()));
>> > >> > }
>> > >> >
>> > >> > String mappedTopic =
>> > >> findExplicitTablePlaceholderTopic(tablePath);
>> > >> > pulsarAdminTool.createTopic(mappedTopic, 1);
>> > >> >
>> > >> > // use pulsar schema to store explicit table information
>> > >> > try {
>> > >> > SchemaInfo schemaInfo =
>> > >> > TableSchemaHelper.generateSchemaInfo(table.toProperties());
>> > >> > pulsarAdminTool.uploadSchema(mappedTopic, schemaInfo);
>> > >> > } catch (Exception e) {
>> > >> > // delete topic if table info cannot be persisted
>> > >> > try {
>> > >> > pulsarAdminTool.deleteTopic(mappedTopic);
>> > >> > } catch (PulsarAdminException ex) {
>> > >> > // do nothing
>> > >> > }
>> > >> > e.printStackTrace();
>> > >> > throw new CatalogException("Can't store table
>> metadata");
>> > >> > }
>> > >> > }
>> > >> > ```
>> > >> >
>> > >> > 2. In
>> `TableSchemaHelper.generateSchemaInfo(table.toProperties());`：
>> > >> > Why must SchemaType.BYTE be used? Is it OK to use SchemaType.JSON?
>> > >> > ```java
>> > >> > public static SchemaInfo generateSchemaInfo(Map<String, String>
>> > >> > properties)
>> > >> > throws JsonProcessingException {
>> > >> > ObjectMapper mapper = new ObjectMapper();
>> > >> > // json
>> > >> > String json = mapper.writeValueAsString(properties);
>> > >> > return SchemaInfoImpl.builder()
>> > >> > .name("flink_table_schema")
>> > >> > //.type(SchemaType.BYTES)
>> > >> > //.schema(mapper.writeValueAsBytes(properties))
>> > >> > // SchemaType.JSON
>> > >> > .type(SchemaType.JSON)
>> > >> > .schema(json.getBytes())
>> > >> > .build();
>> > >> > }
>> > >> > ```
>> > >> >
>> > >> > 3. Sorry, I'm a newbie. Can an experienced developer help with
>> this?
>> > >> Thanks!
>> > >> >
>> > >> >
>> > >> > Thanks,
>> > >> > sinan
>> > >> >
>> > >> >
>> > >> > Enrico Olivelli <eo...@gmail.com> 于2023年4月14日周五 22:50写道：
>> > >> >
>> > >> > > Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet
>> > >> > > <bo...@gmail.com> ha scritto:
>> > >> > > >
>> > >> > > > The change was not reverted for 3.0.0-RC so in the current
>> state the
>> > >> > > > Flink Pulsar catalog won't work with Pulsar 3.0.
>> > >> > > > To give more details, Flink Pulsar stores Flink Table metadata
>> in
>> > >> the
>> > >> > > > metadata of BYTES schema topics (those topics don't have
>> messages,
>> > >> > > > only the schema metadata of these topics is used).
>> > >> > >
>> > >> > > I think that we should not introduce this breaking change, it
>> will be
>> > >> > > a big pain for the users in the Flink ecosystem
>> > >> > >
>> > >> > > Enrico
>> > >> > >
>> > >> > > >
>> > >> > > > Le mer. 29 mars 2023 à 09:54, Yufan Sheng <sy...@gmail.com> a
>> > >> écrit :
>> > >> > > > >
>> > >> > > > > Hi SiNan,
>> > >> > > > >
>> > >> > > > > In the flink world, we don't always rely on the schema
>> information
>> > >> > > > > provided by Pulsar or other connector systems. Flink
>> application
>> > >> has
>> > >> > > > > its own (de)serialization schema logic, which treats the
>> messages
>> > >> only
>> > >> > > > > in a binary format like a byte array.
>> > >> > > > >
>> > >> > > > > In flink-connector-pulsar, we only use the schema when the
>> users
>> > >> want
>> > >> > > > > to do some evolution check. Otherwise, we will only send
>> messages
>> > >> in
>> > >> > > > > BYTES schema.
>> > >> > > > >
>> > >> > > > > On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <
>> > >> liusinan1998@gmail.com>
>> > >> > > wrote:
>> > >> > > > > >
>> > >> > > > > > Hi yufan.
>> > >> > > > > > Can you describe a bit the usage scenario of byte schema in
>> > >> > > > > > flink-connector-pulsa?
>> > >> > > > > >
>> > >> > > > > >
>> > >> > > > > > Thanks,
>> > >> > > > > > sinan
>> > >> > > > > >
>> > >> > > > > > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
>> > >> > > > > >
>> > >> > > > > > > As the flink-connector-pulsar developer, I don't want to
>> > >> disable
>> > >> > > the
>> > >> > > > > > > BYTES schema upload. In my opinion, using BYTES schema
>> means
>> > >> the
>> > >> > > users
>> > >> > > > > > > want to bypass the schema check and handle the schema
>> > >> validation by
>> > >> > > > > > > themselves.
>> > >> > > > > > >
>> > >> > > > > > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <
>> > >> liusinan1998@gmail.com>
>> > >> > > wrote:
>> > >> > > > > > > >
>> > >> > > > > > > > Hi, everyone.
>> > >> > > > > > > > When a user uploads bytes schema. We can warn the user
>> and
>> > >> skip
>> > >> > > uploading
>> > >> > > > > > > > bytes schema.
>> > >> > > > > > > > Also check to see if the topic has a schema other than
>> > >> bytes.
>> > >> > > > > > > > 1. If yes, warn the user that it is not necessary to
>> upload
>> > >> > > bytes schema.
>> > >> > > > > > > > You can subscribe to a topic using bytes schema.
>> > >> > > > > > > > 2. If there is no schema, warn the user that the topic
>> does
>> > >> not
>> > >> > > have a
>> > >> > > > > > > > schema. The default is bytes schema, and there is no
>> need to
>> > >> > > upload it.
>> > >> > > > > > > > Rather than simply throwing an exception rejecting the
>> > >> upload
>> > >> > > bytes
>> > >> > > > > > > schema.
>> > >> > > > > > > >
>> > >> > > > > > > >
>> > >> > > > > > > > Thanks,
>> > >> > > > > > > > sinan
>> > >> > > > > > > >
>> > >> > > > > > > >
>> > >> > > > > > > > Christophe Bornet <bo...@gmail.com> 于
>> 2023年3月28日周二
>> > >> > > 上午1:15写道：
>> > >> > > > > > > >
>> > >> > > > > > > > > This change broke the Flink SQL Pulsar connector:
>> > >> > > > > > > > > https://github.com/streamnative/flink/issues/270
>> > >> > > > > > > > > So I propose to revert it.
>> > >> > > > > > > > >
>> > >> > > > > > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <
>> > >> > > labuladong@foxmail.com> a
>> > >> > > > > > > écrit :
>> > >> > > > > > > > > >
>> > >> > > > > > > > > > Hi pulsar community,
>> > >> > > > > > > > > >
>> > >> > > > > > > > > >
>> > >> > > > > > > > > > I'd like to discuss the behavior of schema
>> uploading,
>> > >> for
>> > >> > > more
>> > >> > > > > > > context
>> > >> > > > > > > > > see https://github.com/apache/pulsar/issues/18825
>> > >> > > > > > > > > >
>> > >> > > > > > > > > >
>> > >> > > > > > > > > > I think that forbidding users to upload `BYTES`
>> schema
>> > >> is a
>> > >> > > > > > > recommended
>> > >> > > > > > > > > way to solve this issue. But this may change the
>> existing
>> > >> > > behavior, so
>> > >> > > > > > > do
>> > >> > > > > > > > > you have any suggestion about this issue?
>> > >> > > > > > > > > >
>> > >> > > > > > > > > >
>> > >> > > > > > > > > > Thanks,
>> > >> > > > > > > > > > Donglai
>> > >> > > > > > > > >
>> > >> > > > > > >
>> > >> > >
>> > >>
>> > >
>>
>
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by PengHui Li <pe...@apache.org>.
> Flink uses the schema to store some kv based properties. If we can
expose all the operations of the topic level metadata store. We can
truly drop the use of uploading the BYTES schema in the Flink
connector.
After 3.0.0, Pulsar will provide the ability to set properties for a topic
https://github.com/apache/pulsar/pull/17238.
Thanks,
Penghui
On Mon, Apr 17, 2023 at 11:44 PM Yufan Sheng <sy...@gmail.com> wrote:
> >I support reverting the PR first and then looking for a long-term
> solution.
>
> Flink uses the schema to store some kv based properties. If we can
> expose all the operations of the topic level metadata store. We can
> truly drop the use of uploading the BYTES schema in the Flink
> connector.
>
> On Mon, Apr 17, 2023 at 12:12 PM PengHui Li <pe...@apache.org> wrote:
> >
> > I'm sorry. I have provided the wrong description of the changes from the
> PR.
> > The PR has changed the server side, so it's hard for users to ask to
> > upgrade
> > the Flink connector if the pulsar server is upgraded.
> >
> > I support reverting the PR first and then looking for a long-term
> solution.
> >
> > Best,
> > Penghui
> >
> > On Mon, Apr 17, 2023 at 10:16 AM PengHui Li <pe...@apache.org> wrote:
> >
> > > > Did we consider
> > > making a call to upload a Bytes schema a no-op?
> > >
> > > It was a BUG that the PR fixed.
> > > You will not be able to get the uploaded schema as expected.
> > > Please take a look at the details from the GitHub issue.
> > >
> > > What is the challenge for the Flink connector now?
> > > The changes only take effect on the client side.
> > > So, the issue will only happen if they use a new connector.
> > > Upgrading the Pulsar server will not make any impaction?
> > > Is it better to fix the Flink connector?
> > > IMO, the Flink connector should not use admin-api
> > > to upload a BYTE schema. It's a redundant operation.
> > > Pulsar will do nothing.
> > >
> > > What do you think about a long-term solution?
> > >
> > > Regards
> > > - Penghui
> > >
> > > On Sat, Apr 15, 2023 at 12:52 AM Michael Marshall <
> mmarshall@apache.org>
> > > wrote:
> > >
> > >> I think the primary point is that unless there is a strict need, we
> > >> shouldn't introduce breaking changes to the implementation. Why did we
> > >> choose to forbid users from uploading a Bytes schema? Did we consider
> > >> making a call to upload a Bytes schema a no-op?
> > >>
> > >> Thanks,
> > >> Michael
> > >>
> > >> On Fri, Apr 14, 2023 at 10:46 AM SiNan Liu <li...@gmail.com>
> > >> wrote:
> > >> >
> > >> > 1. I don't know much about flink, but what I see here is that you
> need
> > >> to
> > >> > save a `ResolvedCatalogTable`, which I see has `CatalogTable`, so
> it is
> > >> > used to record the metadata information of the table.
> > >> > **In
> > >> >
> > >>
> org.apache.flink.connector.pulsar.table.catalog.impl.PulsarCatalogSupport#createTable**
> > >> > ```java
> > >> > @Override
> > >> > public void createTable(ObjectPath tablePath,
> ResolvedCatalogTable
> > >> > table)
> > >> > throws PulsarAdminException {
> > >> > // only allow creating table in explict database, the topic
> is
> > >> used
> > >> > to save table
> > >> > // information
> > >> > if (!isExplicitDatabase(tablePath.getDatabaseName())) {
> > >> > throw new CatalogException(
> > >> > String.format(
> > >> > "Can't create explict table under pulsar
> > >> > tenant/namespace: %s because it's a native database",
> > >> > tablePath.getDatabaseName()));
> > >> > }
> > >> >
> > >> > String mappedTopic =
> > >> findExplicitTablePlaceholderTopic(tablePath);
> > >> > pulsarAdminTool.createTopic(mappedTopic, 1);
> > >> >
> > >> > // use pulsar schema to store explicit table information
> > >> > try {
> > >> > SchemaInfo schemaInfo =
> > >> > TableSchemaHelper.generateSchemaInfo(table.toProperties());
> > >> > pulsarAdminTool.uploadSchema(mappedTopic, schemaInfo);
> > >> > } catch (Exception e) {
> > >> > // delete topic if table info cannot be persisted
> > >> > try {
> > >> > pulsarAdminTool.deleteTopic(mappedTopic);
> > >> > } catch (PulsarAdminException ex) {
> > >> > // do nothing
> > >> > }
> > >> > e.printStackTrace();
> > >> > throw new CatalogException("Can't store table
> metadata");
> > >> > }
> > >> > }
> > >> > ```
> > >> >
> > >> > 2. In `TableSchemaHelper.generateSchemaInfo(table.toProperties());`：
> > >> > Why must SchemaType.BYTE be used? Is it OK to use SchemaType.JSON?
> > >> > ```java
> > >> > public static SchemaInfo generateSchemaInfo(Map<String, String>
> > >> > properties)
> > >> > throws JsonProcessingException {
> > >> > ObjectMapper mapper = new ObjectMapper();
> > >> > // json
> > >> > String json = mapper.writeValueAsString(properties);
> > >> > return SchemaInfoImpl.builder()
> > >> > .name("flink_table_schema")
> > >> > //.type(SchemaType.BYTES)
> > >> > //.schema(mapper.writeValueAsBytes(properties))
> > >> > // SchemaType.JSON
> > >> > .type(SchemaType.JSON)
> > >> > .schema(json.getBytes())
> > >> > .build();
> > >> > }
> > >> > ```
> > >> >
> > >> > 3. Sorry, I'm a newbie. Can an experienced developer help with this?
> > >> Thanks!
> > >> >
> > >> >
> > >> > Thanks,
> > >> > sinan
> > >> >
> > >> >
> > >> > Enrico Olivelli <eo...@gmail.com> 于2023年4月14日周五 22:50写道：
> > >> >
> > >> > > Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet
> > >> > > <bo...@gmail.com> ha scritto:
> > >> > > >
> > >> > > > The change was not reverted for 3.0.0-RC so in the current
> state the
> > >> > > > Flink Pulsar catalog won't work with Pulsar 3.0.
> > >> > > > To give more details, Flink Pulsar stores Flink Table metadata
> in
> > >> the
> > >> > > > metadata of BYTES schema topics (those topics don't have
> messages,
> > >> > > > only the schema metadata of these topics is used).
> > >> > >
> > >> > > I think that we should not introduce this breaking change, it
> will be
> > >> > > a big pain for the users in the Flink ecosystem
> > >> > >
> > >> > > Enrico
> > >> > >
> > >> > > >
> > >> > > > Le mer. 29 mars 2023 à 09:54, Yufan Sheng <sy...@gmail.com> a
> > >> écrit :
> > >> > > > >
> > >> > > > > Hi SiNan,
> > >> > > > >
> > >> > > > > In the flink world, we don't always rely on the schema
> information
> > >> > > > > provided by Pulsar or other connector systems. Flink
> application
> > >> has
> > >> > > > > its own (de)serialization schema logic, which treats the
> messages
> > >> only
> > >> > > > > in a binary format like a byte array.
> > >> > > > >
> > >> > > > > In flink-connector-pulsar, we only use the schema when the
> users
> > >> want
> > >> > > > > to do some evolution check. Otherwise, we will only send
> messages
> > >> in
> > >> > > > > BYTES schema.
> > >> > > > >
> > >> > > > > On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <
> > >> liusinan1998@gmail.com>
> > >> > > wrote:
> > >> > > > > >
> > >> > > > > > Hi yufan.
> > >> > > > > > Can you describe a bit the usage scenario of byte schema in
> > >> > > > > > flink-connector-pulsa?
> > >> > > > > >
> > >> > > > > >
> > >> > > > > > Thanks,
> > >> > > > > > sinan
> > >> > > > > >
> > >> > > > > > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
> > >> > > > > >
> > >> > > > > > > As the flink-connector-pulsar developer, I don't want to
> > >> disable
> > >> > > the
> > >> > > > > > > BYTES schema upload. In my opinion, using BYTES schema
> means
> > >> the
> > >> > > users
> > >> > > > > > > want to bypass the schema check and handle the schema
> > >> validation by
> > >> > > > > > > themselves.
> > >> > > > > > >
> > >> > > > > > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <
> > >> liusinan1998@gmail.com>
> > >> > > wrote:
> > >> > > > > > > >
> > >> > > > > > > > Hi, everyone.
> > >> > > > > > > > When a user uploads bytes schema. We can warn the user
> and
> > >> skip
> > >> > > uploading
> > >> > > > > > > > bytes schema.
> > >> > > > > > > > Also check to see if the topic has a schema other than
> > >> bytes.
> > >> > > > > > > > 1. If yes, warn the user that it is not necessary to
> upload
> > >> > > bytes schema.
> > >> > > > > > > > You can subscribe to a topic using bytes schema.
> > >> > > > > > > > 2. If there is no schema, warn the user that the topic
> does
> > >> not
> > >> > > have a
> > >> > > > > > > > schema. The default is bytes schema, and there is no
> need to
> > >> > > upload it.
> > >> > > > > > > > Rather than simply throwing an exception rejecting the
> > >> upload
> > >> > > bytes
> > >> > > > > > > schema.
> > >> > > > > > > >
> > >> > > > > > > >
> > >> > > > > > > > Thanks,
> > >> > > > > > > > sinan
> > >> > > > > > > >
> > >> > > > > > > >
> > >> > > > > > > > Christophe Bornet <bo...@gmail.com> 于
> 2023年3月28日周二
> > >> > > 上午1:15写道：
> > >> > > > > > > >
> > >> > > > > > > > > This change broke the Flink SQL Pulsar connector:
> > >> > > > > > > > > https://github.com/streamnative/flink/issues/270
> > >> > > > > > > > > So I propose to revert it.
> > >> > > > > > > > >
> > >> > > > > > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <
> > >> > > labuladong@foxmail.com> a
> > >> > > > > > > écrit :
> > >> > > > > > > > > >
> > >> > > > > > > > > > Hi pulsar community,
> > >> > > > > > > > > >
> > >> > > > > > > > > >
> > >> > > > > > > > > > I'd like to discuss the behavior of schema
> uploading,
> > >> for
> > >> > > more
> > >> > > > > > > context
> > >> > > > > > > > > see https://github.com/apache/pulsar/issues/18825
> > >> > > > > > > > > >
> > >> > > > > > > > > >
> > >> > > > > > > > > > I think that forbidding users to upload `BYTES`
> schema
> > >> is a
> > >> > > > > > > recommended
> > >> > > > > > > > > way to solve this issue. But this may change the
> existing
> > >> > > behavior, so
> > >> > > > > > > do
> > >> > > > > > > > > you have any suggestion about this issue?
> > >> > > > > > > > > >
> > >> > > > > > > > > >
> > >> > > > > > > > > > Thanks,
> > >> > > > > > > > > > Donglai
> > >> > > > > > > > >
> > >> > > > > > >
> > >> > >
> > >>
> > >
>
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by Yufan Sheng <sy...@gmail.com>.
>I support reverting the PR first and then looking for a long-term solution.
Flink uses the schema to store some kv based properties. If we can
expose all the operations of the topic level metadata store. We can
truly drop the use of uploading the BYTES schema in the Flink
connector.
On Mon, Apr 17, 2023 at 12:12 PM PengHui Li <pe...@apache.org> wrote:
>
> I'm sorry. I have provided the wrong description of the changes from the PR.
> The PR has changed the server side, so it's hard for users to ask to
> upgrade
> the Flink connector if the pulsar server is upgraded.
>
> I support reverting the PR first and then looking for a long-term solution.
>
> Best,
> Penghui
>
> On Mon, Apr 17, 2023 at 10:16 AM PengHui Li <pe...@apache.org> wrote:
>
> > > Did we consider
> > making a call to upload a Bytes schema a no-op?
> >
> > It was a BUG that the PR fixed.
> > You will not be able to get the uploaded schema as expected.
> > Please take a look at the details from the GitHub issue.
> >
> > What is the challenge for the Flink connector now?
> > The changes only take effect on the client side.
> > So, the issue will only happen if they use a new connector.
> > Upgrading the Pulsar server will not make any impaction?
> > Is it better to fix the Flink connector?
> > IMO, the Flink connector should not use admin-api
> > to upload a BYTE schema. It's a redundant operation.
> > Pulsar will do nothing.
> >
> > What do you think about a long-term solution?
> >
> > Regards
> > - Penghui
> >
> > On Sat, Apr 15, 2023 at 12:52 AM Michael Marshall <mm...@apache.org>
> > wrote:
> >
> >> I think the primary point is that unless there is a strict need, we
> >> shouldn't introduce breaking changes to the implementation. Why did we
> >> choose to forbid users from uploading a Bytes schema? Did we consider
> >> making a call to upload a Bytes schema a no-op?
> >>
> >> Thanks,
> >> Michael
> >>
> >> On Fri, Apr 14, 2023 at 10:46 AM SiNan Liu <li...@gmail.com>
> >> wrote:
> >> >
> >> > 1. I don't know much about flink, but what I see here is that you need
> >> to
> >> > save a `ResolvedCatalogTable`, which I see has `CatalogTable`, so it is
> >> > used to record the metadata information of the table.
> >> > **In
> >> >
> >> org.apache.flink.connector.pulsar.table.catalog.impl.PulsarCatalogSupport#createTable**
> >> > ```java
> >> > @Override
> >> > public void createTable(ObjectPath tablePath, ResolvedCatalogTable
> >> > table)
> >> > throws PulsarAdminException {
> >> > // only allow creating table in explict database, the topic is
> >> used
> >> > to save table
> >> > // information
> >> > if (!isExplicitDatabase(tablePath.getDatabaseName())) {
> >> > throw new CatalogException(
> >> > String.format(
> >> > "Can't create explict table under pulsar
> >> > tenant/namespace: %s because it's a native database",
> >> > tablePath.getDatabaseName()));
> >> > }
> >> >
> >> > String mappedTopic =
> >> findExplicitTablePlaceholderTopic(tablePath);
> >> > pulsarAdminTool.createTopic(mappedTopic, 1);
> >> >
> >> > // use pulsar schema to store explicit table information
> >> > try {
> >> > SchemaInfo schemaInfo =
> >> > TableSchemaHelper.generateSchemaInfo(table.toProperties());
> >> > pulsarAdminTool.uploadSchema(mappedTopic, schemaInfo);
> >> > } catch (Exception e) {
> >> > // delete topic if table info cannot be persisted
> >> > try {
> >> > pulsarAdminTool.deleteTopic(mappedTopic);
> >> > } catch (PulsarAdminException ex) {
> >> > // do nothing
> >> > }
> >> > e.printStackTrace();
> >> > throw new CatalogException("Can't store table metadata");
> >> > }
> >> > }
> >> > ```
> >> >
> >> > 2. In `TableSchemaHelper.generateSchemaInfo(table.toProperties());`：
> >> > Why must SchemaType.BYTE be used? Is it OK to use SchemaType.JSON?
> >> > ```java
> >> > public static SchemaInfo generateSchemaInfo(Map<String, String>
> >> > properties)
> >> > throws JsonProcessingException {
> >> > ObjectMapper mapper = new ObjectMapper();
> >> > // json
> >> > String json = mapper.writeValueAsString(properties);
> >> > return SchemaInfoImpl.builder()
> >> > .name("flink_table_schema")
> >> > //.type(SchemaType.BYTES)
> >> > //.schema(mapper.writeValueAsBytes(properties))
> >> > // SchemaType.JSON
> >> > .type(SchemaType.JSON)
> >> > .schema(json.getBytes())
> >> > .build();
> >> > }
> >> > ```
> >> >
> >> > 3. Sorry, I'm a newbie. Can an experienced developer help with this?
> >> Thanks!
> >> >
> >> >
> >> > Thanks,
> >> > sinan
> >> >
> >> >
> >> > Enrico Olivelli <eo...@gmail.com> 于2023年4月14日周五 22:50写道：
> >> >
> >> > > Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet
> >> > > <bo...@gmail.com> ha scritto:
> >> > > >
> >> > > > The change was not reverted for 3.0.0-RC so in the current state the
> >> > > > Flink Pulsar catalog won't work with Pulsar 3.0.
> >> > > > To give more details, Flink Pulsar stores Flink Table metadata in
> >> the
> >> > > > metadata of BYTES schema topics (those topics don't have messages,
> >> > > > only the schema metadata of these topics is used).
> >> > >
> >> > > I think that we should not introduce this breaking change, it will be
> >> > > a big pain for the users in the Flink ecosystem
> >> > >
> >> > > Enrico
> >> > >
> >> > > >
> >> > > > Le mer. 29 mars 2023 à 09:54, Yufan Sheng <sy...@gmail.com> a
> >> écrit :
> >> > > > >
> >> > > > > Hi SiNan,
> >> > > > >
> >> > > > > In the flink world, we don't always rely on the schema information
> >> > > > > provided by Pulsar or other connector systems. Flink application
> >> has
> >> > > > > its own (de)serialization schema logic, which treats the messages
> >> only
> >> > > > > in a binary format like a byte array.
> >> > > > >
> >> > > > > In flink-connector-pulsar, we only use the schema when the users
> >> want
> >> > > > > to do some evolution check. Otherwise, we will only send messages
> >> in
> >> > > > > BYTES schema.
> >> > > > >
> >> > > > > On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <
> >> liusinan1998@gmail.com>
> >> > > wrote:
> >> > > > > >
> >> > > > > > Hi yufan.
> >> > > > > > Can you describe a bit the usage scenario of byte schema in
> >> > > > > > flink-connector-pulsa?
> >> > > > > >
> >> > > > > >
> >> > > > > > Thanks,
> >> > > > > > sinan
> >> > > > > >
> >> > > > > > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
> >> > > > > >
> >> > > > > > > As the flink-connector-pulsar developer, I don't want to
> >> disable
> >> > > the
> >> > > > > > > BYTES schema upload. In my opinion, using BYTES schema means
> >> the
> >> > > users
> >> > > > > > > want to bypass the schema check and handle the schema
> >> validation by
> >> > > > > > > themselves.
> >> > > > > > >
> >> > > > > > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <
> >> liusinan1998@gmail.com>
> >> > > wrote:
> >> > > > > > > >
> >> > > > > > > > Hi, everyone.
> >> > > > > > > > When a user uploads bytes schema. We can warn the user and
> >> skip
> >> > > uploading
> >> > > > > > > > bytes schema.
> >> > > > > > > > Also check to see if the topic has a schema other than
> >> bytes.
> >> > > > > > > > 1. If yes, warn the user that it is not necessary to upload
> >> > > bytes schema.
> >> > > > > > > > You can subscribe to a topic using bytes schema.
> >> > > > > > > > 2. If there is no schema, warn the user that the topic does
> >> not
> >> > > have a
> >> > > > > > > > schema. The default is bytes schema, and there is no need to
> >> > > upload it.
> >> > > > > > > > Rather than simply throwing an exception rejecting the
> >> upload
> >> > > bytes
> >> > > > > > > schema.
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > > Thanks,
> >> > > > > > > > sinan
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > > Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二
> >> > > 上午1:15写道：
> >> > > > > > > >
> >> > > > > > > > > This change broke the Flink SQL Pulsar connector:
> >> > > > > > > > > https://github.com/streamnative/flink/issues/270
> >> > > > > > > > > So I propose to revert it.
> >> > > > > > > > >
> >> > > > > > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <
> >> > > labuladong@foxmail.com> a
> >> > > > > > > écrit :
> >> > > > > > > > > >
> >> > > > > > > > > > Hi pulsar community,
> >> > > > > > > > > >
> >> > > > > > > > > >
> >> > > > > > > > > > I'd like to discuss the behavior of schema uploading,
> >> for
> >> > > more
> >> > > > > > > context
> >> > > > > > > > > see https://github.com/apache/pulsar/issues/18825
> >> > > > > > > > > >
> >> > > > > > > > > >
> >> > > > > > > > > > I think that forbidding users to upload `BYTES` schema
> >> is a
> >> > > > > > > recommended
> >> > > > > > > > > way to solve this issue. But this may change the existing
> >> > > behavior, so
> >> > > > > > > do
> >> > > > > > > > > you have any suggestion about this issue?
> >> > > > > > > > > >
> >> > > > > > > > > >
> >> > > > > > > > > > Thanks,
> >> > > > > > > > > > Donglai
> >> > > > > > > > >
> >> > > > > > >
> >> > >
> >>
> >
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by PengHui Li <pe...@apache.org>.
I'm sorry. I have provided the wrong description of the changes from the PR.
The PR has changed the server side, so it's hard for users to ask to
upgrade
the Flink connector if the pulsar server is upgraded.
I support reverting the PR first and then looking for a long-term solution.
Best,
Penghui
On Mon, Apr 17, 2023 at 10:16 AM PengHui Li <pe...@apache.org> wrote:
> > Did we consider
> making a call to upload a Bytes schema a no-op?
>
> It was a BUG that the PR fixed.
> You will not be able to get the uploaded schema as expected.
> Please take a look at the details from the GitHub issue.
>
> What is the challenge for the Flink connector now?
> The changes only take effect on the client side.
> So, the issue will only happen if they use a new connector.
> Upgrading the Pulsar server will not make any impaction?
> Is it better to fix the Flink connector?
> IMO, the Flink connector should not use admin-api
> to upload a BYTE schema. It's a redundant operation.
> Pulsar will do nothing.
>
> What do you think about a long-term solution?
>
> Regards
> - Penghui
>
> On Sat, Apr 15, 2023 at 12:52 AM Michael Marshall <mm...@apache.org>
> wrote:
>
>> I think the primary point is that unless there is a strict need, we
>> shouldn't introduce breaking changes to the implementation. Why did we
>> choose to forbid users from uploading a Bytes schema? Did we consider
>> making a call to upload a Bytes schema a no-op?
>>
>> Thanks,
>> Michael
>>
>> On Fri, Apr 14, 2023 at 10:46 AM SiNan Liu <li...@gmail.com>
>> wrote:
>> >
>> > 1. I don't know much about flink, but what I see here is that you need
>> to
>> > save a `ResolvedCatalogTable`, which I see has `CatalogTable`, so it is
>> > used to record the metadata information of the table.
>> > **In
>> >
>> org.apache.flink.connector.pulsar.table.catalog.impl.PulsarCatalogSupport#createTable**
>> > ```java
>> > @Override
>> > public void createTable(ObjectPath tablePath, ResolvedCatalogTable
>> > table)
>> > throws PulsarAdminException {
>> > // only allow creating table in explict database, the topic is
>> used
>> > to save table
>> > // information
>> > if (!isExplicitDatabase(tablePath.getDatabaseName())) {
>> > throw new CatalogException(
>> > String.format(
>> > "Can't create explict table under pulsar
>> > tenant/namespace: %s because it's a native database",
>> > tablePath.getDatabaseName()));
>> > }
>> >
>> > String mappedTopic =
>> findExplicitTablePlaceholderTopic(tablePath);
>> > pulsarAdminTool.createTopic(mappedTopic, 1);
>> >
>> > // use pulsar schema to store explicit table information
>> > try {
>> > SchemaInfo schemaInfo =
>> > TableSchemaHelper.generateSchemaInfo(table.toProperties());
>> > pulsarAdminTool.uploadSchema(mappedTopic, schemaInfo);
>> > } catch (Exception e) {
>> > // delete topic if table info cannot be persisted
>> > try {
>> > pulsarAdminTool.deleteTopic(mappedTopic);
>> > } catch (PulsarAdminException ex) {
>> > // do nothing
>> > }
>> > e.printStackTrace();
>> > throw new CatalogException("Can't store table metadata");
>> > }
>> > }
>> > ```
>> >
>> > 2. In `TableSchemaHelper.generateSchemaInfo(table.toProperties());`：
>> > Why must SchemaType.BYTE be used? Is it OK to use SchemaType.JSON?
>> > ```java
>> > public static SchemaInfo generateSchemaInfo(Map<String, String>
>> > properties)
>> > throws JsonProcessingException {
>> > ObjectMapper mapper = new ObjectMapper();
>> > // json
>> > String json = mapper.writeValueAsString(properties);
>> > return SchemaInfoImpl.builder()
>> > .name("flink_table_schema")
>> > //.type(SchemaType.BYTES)
>> > //.schema(mapper.writeValueAsBytes(properties))
>> > // SchemaType.JSON
>> > .type(SchemaType.JSON)
>> > .schema(json.getBytes())
>> > .build();
>> > }
>> > ```
>> >
>> > 3. Sorry, I'm a newbie. Can an experienced developer help with this?
>> Thanks!
>> >
>> >
>> > Thanks,
>> > sinan
>> >
>> >
>> > Enrico Olivelli <eo...@gmail.com> 于2023年4月14日周五 22:50写道：
>> >
>> > > Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet
>> > > <bo...@gmail.com> ha scritto:
>> > > >
>> > > > The change was not reverted for 3.0.0-RC so in the current state the
>> > > > Flink Pulsar catalog won't work with Pulsar 3.0.
>> > > > To give more details, Flink Pulsar stores Flink Table metadata in
>> the
>> > > > metadata of BYTES schema topics (those topics don't have messages,
>> > > > only the schema metadata of these topics is used).
>> > >
>> > > I think that we should not introduce this breaking change, it will be
>> > > a big pain for the users in the Flink ecosystem
>> > >
>> > > Enrico
>> > >
>> > > >
>> > > > Le mer. 29 mars 2023 à 09:54, Yufan Sheng <sy...@gmail.com> a
>> écrit :
>> > > > >
>> > > > > Hi SiNan,
>> > > > >
>> > > > > In the flink world, we don't always rely on the schema information
>> > > > > provided by Pulsar or other connector systems. Flink application
>> has
>> > > > > its own (de)serialization schema logic, which treats the messages
>> only
>> > > > > in a binary format like a byte array.
>> > > > >
>> > > > > In flink-connector-pulsar, we only use the schema when the users
>> want
>> > > > > to do some evolution check. Otherwise, we will only send messages
>> in
>> > > > > BYTES schema.
>> > > > >
>> > > > > On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <
>> liusinan1998@gmail.com>
>> > > wrote:
>> > > > > >
>> > > > > > Hi yufan.
>> > > > > > Can you describe a bit the usage scenario of byte schema in
>> > > > > > flink-connector-pulsa?
>> > > > > >
>> > > > > >
>> > > > > > Thanks,
>> > > > > > sinan
>> > > > > >
>> > > > > > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
>> > > > > >
>> > > > > > > As the flink-connector-pulsar developer, I don't want to
>> disable
>> > > the
>> > > > > > > BYTES schema upload. In my opinion, using BYTES schema means
>> the
>> > > users
>> > > > > > > want to bypass the schema check and handle the schema
>> validation by
>> > > > > > > themselves.
>> > > > > > >
>> > > > > > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <
>> liusinan1998@gmail.com>
>> > > wrote:
>> > > > > > > >
>> > > > > > > > Hi, everyone.
>> > > > > > > > When a user uploads bytes schema. We can warn the user and
>> skip
>> > > uploading
>> > > > > > > > bytes schema.
>> > > > > > > > Also check to see if the topic has a schema other than
>> bytes.
>> > > > > > > > 1. If yes, warn the user that it is not necessary to upload
>> > > bytes schema.
>> > > > > > > > You can subscribe to a topic using bytes schema.
>> > > > > > > > 2. If there is no schema, warn the user that the topic does
>> not
>> > > have a
>> > > > > > > > schema. The default is bytes schema, and there is no need to
>> > > upload it.
>> > > > > > > > Rather than simply throwing an exception rejecting the
>> upload
>> > > bytes
>> > > > > > > schema.
>> > > > > > > >
>> > > > > > > >
>> > > > > > > > Thanks,
>> > > > > > > > sinan
>> > > > > > > >
>> > > > > > > >
>> > > > > > > > Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二
>> > > 上午1:15写道：
>> > > > > > > >
>> > > > > > > > > This change broke the Flink SQL Pulsar connector:
>> > > > > > > > > https://github.com/streamnative/flink/issues/270
>> > > > > > > > > So I propose to revert it.
>> > > > > > > > >
>> > > > > > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <
>> > > labuladong@foxmail.com> a
>> > > > > > > écrit :
>> > > > > > > > > >
>> > > > > > > > > > Hi pulsar community,
>> > > > > > > > > >
>> > > > > > > > > >
>> > > > > > > > > > I'd like to discuss the behavior of schema uploading,
>> for
>> > > more
>> > > > > > > context
>> > > > > > > > > see https://github.com/apache/pulsar/issues/18825
>> > > > > > > > > >
>> > > > > > > > > >
>> > > > > > > > > > I think that forbidding users to upload `BYTES` schema
>> is a
>> > > > > > > recommended
>> > > > > > > > > way to solve this issue. But this may change the existing
>> > > behavior, so
>> > > > > > > do
>> > > > > > > > > you have any suggestion about this issue?
>> > > > > > > > > >
>> > > > > > > > > >
>> > > > > > > > > > Thanks,
>> > > > > > > > > > Donglai
>> > > > > > > > >
>> > > > > > >
>> > >
>>
>
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by PengHui Li <pe...@apache.org>.
> Did we consider
making a call to upload a Bytes schema a no-op?
It was a BUG that the PR fixed.
You will not be able to get the uploaded schema as expected.
Please take a look at the details from the GitHub issue.
What is the challenge for the Flink connector now?
The changes only take effect on the client side.
So, the issue will only happen if they use a new connector.
Upgrading the Pulsar server will not make any impaction?
Is it better to fix the Flink connector?
IMO, the Flink connector should not use admin-api
to upload a BYTE schema. It's a redundant operation.
Pulsar will do nothing.
What do you think about a long-term solution?
Regards
- Penghui
On Sat, Apr 15, 2023 at 12:52 AM Michael Marshall <mm...@apache.org>
wrote:
> I think the primary point is that unless there is a strict need, we
> shouldn't introduce breaking changes to the implementation. Why did we
> choose to forbid users from uploading a Bytes schema? Did we consider
> making a call to upload a Bytes schema a no-op?
>
> Thanks,
> Michael
>
> On Fri, Apr 14, 2023 at 10:46 AM SiNan Liu <li...@gmail.com> wrote:
> >
> > 1. I don't know much about flink, but what I see here is that you need to
> > save a `ResolvedCatalogTable`, which I see has `CatalogTable`, so it is
> > used to record the metadata information of the table.
> > **In
> >
> org.apache.flink.connector.pulsar.table.catalog.impl.PulsarCatalogSupport#createTable**
> > ```java
> > @Override
> > public void createTable(ObjectPath tablePath, ResolvedCatalogTable
> > table)
> > throws PulsarAdminException {
> > // only allow creating table in explict database, the topic is
> used
> > to save table
> > // information
> > if (!isExplicitDatabase(tablePath.getDatabaseName())) {
> > throw new CatalogException(
> > String.format(
> > "Can't create explict table under pulsar
> > tenant/namespace: %s because it's a native database",
> > tablePath.getDatabaseName()));
> > }
> >
> > String mappedTopic =
> findExplicitTablePlaceholderTopic(tablePath);
> > pulsarAdminTool.createTopic(mappedTopic, 1);
> >
> > // use pulsar schema to store explicit table information
> > try {
> > SchemaInfo schemaInfo =
> > TableSchemaHelper.generateSchemaInfo(table.toProperties());
> > pulsarAdminTool.uploadSchema(mappedTopic, schemaInfo);
> > } catch (Exception e) {
> > // delete topic if table info cannot be persisted
> > try {
> > pulsarAdminTool.deleteTopic(mappedTopic);
> > } catch (PulsarAdminException ex) {
> > // do nothing
> > }
> > e.printStackTrace();
> > throw new CatalogException("Can't store table metadata");
> > }
> > }
> > ```
> >
> > 2. In `TableSchemaHelper.generateSchemaInfo(table.toProperties());`：
> > Why must SchemaType.BYTE be used? Is it OK to use SchemaType.JSON?
> > ```java
> > public static SchemaInfo generateSchemaInfo(Map<String, String>
> > properties)
> > throws JsonProcessingException {
> > ObjectMapper mapper = new ObjectMapper();
> > // json
> > String json = mapper.writeValueAsString(properties);
> > return SchemaInfoImpl.builder()
> > .name("flink_table_schema")
> > //.type(SchemaType.BYTES)
> > //.schema(mapper.writeValueAsBytes(properties))
> > // SchemaType.JSON
> > .type(SchemaType.JSON)
> > .schema(json.getBytes())
> > .build();
> > }
> > ```
> >
> > 3. Sorry, I'm a newbie. Can an experienced developer help with this?
> Thanks!
> >
> >
> > Thanks,
> > sinan
> >
> >
> > Enrico Olivelli <eo...@gmail.com> 于2023年4月14日周五 22:50写道：
> >
> > > Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet
> > > <bo...@gmail.com> ha scritto:
> > > >
> > > > The change was not reverted for 3.0.0-RC so in the current state the
> > > > Flink Pulsar catalog won't work with Pulsar 3.0.
> > > > To give more details, Flink Pulsar stores Flink Table metadata in the
> > > > metadata of BYTES schema topics (those topics don't have messages,
> > > > only the schema metadata of these topics is used).
> > >
> > > I think that we should not introduce this breaking change, it will be
> > > a big pain for the users in the Flink ecosystem
> > >
> > > Enrico
> > >
> > > >
> > > > Le mer. 29 mars 2023 à 09:54, Yufan Sheng <sy...@gmail.com> a
> écrit :
> > > > >
> > > > > Hi SiNan,
> > > > >
> > > > > In the flink world, we don't always rely on the schema information
> > > > > provided by Pulsar or other connector systems. Flink application
> has
> > > > > its own (de)serialization schema logic, which treats the messages
> only
> > > > > in a binary format like a byte array.
> > > > >
> > > > > In flink-connector-pulsar, we only use the schema when the users
> want
> > > > > to do some evolution check. Otherwise, we will only send messages
> in
> > > > > BYTES schema.
> > > > >
> > > > > On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <liusinan1998@gmail.com
> >
> > > wrote:
> > > > > >
> > > > > > Hi yufan.
> > > > > > Can you describe a bit the usage scenario of byte schema in
> > > > > > flink-connector-pulsa?
> > > > > >
> > > > > >
> > > > > > Thanks,
> > > > > > sinan
> > > > > >
> > > > > > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
> > > > > >
> > > > > > > As the flink-connector-pulsar developer, I don't want to
> disable
> > > the
> > > > > > > BYTES schema upload. In my opinion, using BYTES schema means
> the
> > > users
> > > > > > > want to bypass the schema check and handle the schema
> validation by
> > > > > > > themselves.
> > > > > > >
> > > > > > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <
> liusinan1998@gmail.com>
> > > wrote:
> > > > > > > >
> > > > > > > > Hi, everyone.
> > > > > > > > When a user uploads bytes schema. We can warn the user and
> skip
> > > uploading
> > > > > > > > bytes schema.
> > > > > > > > Also check to see if the topic has a schema other than bytes.
> > > > > > > > 1. If yes, warn the user that it is not necessary to upload
> > > bytes schema.
> > > > > > > > You can subscribe to a topic using bytes schema.
> > > > > > > > 2. If there is no schema, warn the user that the topic does
> not
> > > have a
> > > > > > > > schema. The default is bytes schema, and there is no need to
> > > upload it.
> > > > > > > > Rather than simply throwing an exception rejecting the upload
> > > bytes
> > > > > > > schema.
> > > > > > > >
> > > > > > > >
> > > > > > > > Thanks,
> > > > > > > > sinan
> > > > > > > >
> > > > > > > >
> > > > > > > > Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二
> > > 上午1:15写道：
> > > > > > > >
> > > > > > > > > This change broke the Flink SQL Pulsar connector:
> > > > > > > > > https://github.com/streamnative/flink/issues/270
> > > > > > > > > So I propose to revert it.
> > > > > > > > >
> > > > > > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <
> > > labuladong@foxmail.com> a
> > > > > > > écrit :
> > > > > > > > > >
> > > > > > > > > > Hi pulsar community,
> > > > > > > > > >
> > > > > > > > > >
> > > > > > > > > > I'd like to discuss the behavior of schema uploading, for
> > > more
> > > > > > > context
> > > > > > > > > see https://github.com/apache/pulsar/issues/18825
> > > > > > > > > >
> > > > > > > > > >
> > > > > > > > > > I think that forbidding users to upload `BYTES` schema
> is a
> > > > > > > recommended
> > > > > > > > > way to solve this issue. But this may change the existing
> > > behavior, so
> > > > > > > do
> > > > > > > > > you have any suggestion about this issue?
> > > > > > > > > >
> > > > > > > > > >
> > > > > > > > > > Thanks,
> > > > > > > > > > Donglai
> > > > > > > > >
> > > > > > >
> > >
>
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by Michael Marshall <mm...@apache.org>.
I think the primary point is that unless there is a strict need, we
shouldn't introduce breaking changes to the implementation. Why did we
choose to forbid users from uploading a Bytes schema? Did we consider
making a call to upload a Bytes schema a no-op?
Thanks,
Michael
On Fri, Apr 14, 2023 at 10:46 AM SiNan Liu <li...@gmail.com> wrote:
>
> 1. I don't know much about flink, but what I see here is that you need to
> save a `ResolvedCatalogTable`, which I see has `CatalogTable`, so it is
> used to record the metadata information of the table.
> **In
> org.apache.flink.connector.pulsar.table.catalog.impl.PulsarCatalogSupport#createTable**
> ```java
> @Override
> public void createTable(ObjectPath tablePath, ResolvedCatalogTable
> table)
> throws PulsarAdminException {
> // only allow creating table in explict database, the topic is used
> to save table
> // information
> if (!isExplicitDatabase(tablePath.getDatabaseName())) {
> throw new CatalogException(
> String.format(
> "Can't create explict table under pulsar
> tenant/namespace: %s because it's a native database",
> tablePath.getDatabaseName()));
> }
>
> String mappedTopic = findExplicitTablePlaceholderTopic(tablePath);
> pulsarAdminTool.createTopic(mappedTopic, 1);
>
> // use pulsar schema to store explicit table information
> try {
> SchemaInfo schemaInfo =
> TableSchemaHelper.generateSchemaInfo(table.toProperties());
> pulsarAdminTool.uploadSchema(mappedTopic, schemaInfo);
> } catch (Exception e) {
> // delete topic if table info cannot be persisted
> try {
> pulsarAdminTool.deleteTopic(mappedTopic);
> } catch (PulsarAdminException ex) {
> // do nothing
> }
> e.printStackTrace();
> throw new CatalogException("Can't store table metadata");
> }
> }
> ```
>
> 2. In `TableSchemaHelper.generateSchemaInfo(table.toProperties());`：
> Why must SchemaType.BYTE be used? Is it OK to use SchemaType.JSON?
> ```java
> public static SchemaInfo generateSchemaInfo(Map<String, String>
> properties)
> throws JsonProcessingException {
> ObjectMapper mapper = new ObjectMapper();
> // json
> String json = mapper.writeValueAsString(properties);
> return SchemaInfoImpl.builder()
> .name("flink_table_schema")
> //.type(SchemaType.BYTES)
> //.schema(mapper.writeValueAsBytes(properties))
> // SchemaType.JSON
> .type(SchemaType.JSON)
> .schema(json.getBytes())
> .build();
> }
> ```
>
> 3. Sorry, I'm a newbie. Can an experienced developer help with this? Thanks!
>
>
> Thanks,
> sinan
>
>
> Enrico Olivelli <eo...@gmail.com> 于2023年4月14日周五 22:50写道：
>
> > Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet
> > <bo...@gmail.com> ha scritto:
> > >
> > > The change was not reverted for 3.0.0-RC so in the current state the
> > > Flink Pulsar catalog won't work with Pulsar 3.0.
> > > To give more details, Flink Pulsar stores Flink Table metadata in the
> > > metadata of BYTES schema topics (those topics don't have messages,
> > > only the schema metadata of these topics is used).
> >
> > I think that we should not introduce this breaking change, it will be
> > a big pain for the users in the Flink ecosystem
> >
> > Enrico
> >
> > >
> > > Le mer. 29 mars 2023 à 09:54, Yufan Sheng <sy...@gmail.com> a écrit :
> > > >
> > > > Hi SiNan,
> > > >
> > > > In the flink world, we don't always rely on the schema information
> > > > provided by Pulsar or other connector systems. Flink application has
> > > > its own (de)serialization schema logic, which treats the messages only
> > > > in a binary format like a byte array.
> > > >
> > > > In flink-connector-pulsar, we only use the schema when the users want
> > > > to do some evolution check. Otherwise, we will only send messages in
> > > > BYTES schema.
> > > >
> > > > On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <li...@gmail.com>
> > wrote:
> > > > >
> > > > > Hi yufan.
> > > > > Can you describe a bit the usage scenario of byte schema in
> > > > > flink-connector-pulsa?
> > > > >
> > > > >
> > > > > Thanks,
> > > > > sinan
> > > > >
> > > > > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
> > > > >
> > > > > > As the flink-connector-pulsar developer, I don't want to disable
> > the
> > > > > > BYTES schema upload. In my opinion, using BYTES schema means the
> > users
> > > > > > want to bypass the schema check and handle the schema validation by
> > > > > > themselves.
> > > > > >
> > > > > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <li...@gmail.com>
> > wrote:
> > > > > > >
> > > > > > > Hi, everyone.
> > > > > > > When a user uploads bytes schema. We can warn the user and skip
> > uploading
> > > > > > > bytes schema.
> > > > > > > Also check to see if the topic has a schema other than bytes.
> > > > > > > 1. If yes, warn the user that it is not necessary to upload
> > bytes schema.
> > > > > > > You can subscribe to a topic using bytes schema.
> > > > > > > 2. If there is no schema, warn the user that the topic does not
> > have a
> > > > > > > schema. The default is bytes schema, and there is no need to
> > upload it.
> > > > > > > Rather than simply throwing an exception rejecting the upload
> > bytes
> > > > > > schema.
> > > > > > >
> > > > > > >
> > > > > > > Thanks,
> > > > > > > sinan
> > > > > > >
> > > > > > >
> > > > > > > Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二
> > 上午1:15写道：
> > > > > > >
> > > > > > > > This change broke the Flink SQL Pulsar connector:
> > > > > > > > https://github.com/streamnative/flink/issues/270
> > > > > > > > So I propose to revert it.
> > > > > > > >
> > > > > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <
> > labuladong@foxmail.com> a
> > > > > > écrit :
> > > > > > > > >
> > > > > > > > > Hi pulsar community,
> > > > > > > > >
> > > > > > > > >
> > > > > > > > > I'd like to discuss the behavior of schema uploading, for
> > more
> > > > > > context
> > > > > > > > see https://github.com/apache/pulsar/issues/18825
> > > > > > > > >
> > > > > > > > >
> > > > > > > > > I think that forbidding users to upload `BYTES` schema is a
> > > > > > recommended
> > > > > > > > way to solve this issue. But this may change the existing
> > behavior, so
> > > > > > do
> > > > > > > > you have any suggestion about this issue?
> > > > > > > > >
> > > > > > > > >
> > > > > > > > > Thanks,
> > > > > > > > > Donglai
> > > > > > > >
> > > > > >
> >
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by SiNan Liu <li...@gmail.com>.
1. I don't know much about flink, but what I see here is that you need to
save a `ResolvedCatalogTable`, which I see has `CatalogTable`, so it is
used to record the metadata information of the table.
**In
org.apache.flink.connector.pulsar.table.catalog.impl.PulsarCatalogSupport#createTable**
```java
@Override
public void createTable(ObjectPath tablePath, ResolvedCatalogTable
table)
throws PulsarAdminException {
// only allow creating table in explict database, the topic is used
to save table
// information
if (!isExplicitDatabase(tablePath.getDatabaseName())) {
throw new CatalogException(
String.format(
"Can't create explict table under pulsar
tenant/namespace: %s because it's a native database",
tablePath.getDatabaseName()));
}
String mappedTopic = findExplicitTablePlaceholderTopic(tablePath);
pulsarAdminTool.createTopic(mappedTopic, 1);
// use pulsar schema to store explicit table information
try {
SchemaInfo schemaInfo =
TableSchemaHelper.generateSchemaInfo(table.toProperties());
pulsarAdminTool.uploadSchema(mappedTopic, schemaInfo);
} catch (Exception e) {
// delete topic if table info cannot be persisted
try {
pulsarAdminTool.deleteTopic(mappedTopic);
} catch (PulsarAdminException ex) {
// do nothing
}
e.printStackTrace();
throw new CatalogException("Can't store table metadata");
}
}
```
2. In `TableSchemaHelper.generateSchemaInfo(table.toProperties());`：
Why must SchemaType.BYTE be used? Is it OK to use SchemaType.JSON?
```java
public static SchemaInfo generateSchemaInfo(Map<String, String>
properties)
throws JsonProcessingException {
ObjectMapper mapper = new ObjectMapper();
// json
String json = mapper.writeValueAsString(properties);
return SchemaInfoImpl.builder()
.name("flink_table_schema")
//.type(SchemaType.BYTES)
//.schema(mapper.writeValueAsBytes(properties))
// SchemaType.JSON
.type(SchemaType.JSON)
.schema(json.getBytes())
.build();
}
```
3. Sorry, I'm a newbie. Can an experienced developer help with this? Thanks!
Thanks,
sinan
Enrico Olivelli <eo...@gmail.com> 于2023年4月14日周五 22:50写道：
> Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet
> <bo...@gmail.com> ha scritto:
> >
> > The change was not reverted for 3.0.0-RC so in the current state the
> > Flink Pulsar catalog won't work with Pulsar 3.0.
> > To give more details, Flink Pulsar stores Flink Table metadata in the
> > metadata of BYTES schema topics (those topics don't have messages,
> > only the schema metadata of these topics is used).
>
> I think that we should not introduce this breaking change, it will be
> a big pain for the users in the Flink ecosystem
>
> Enrico
>
> >
> > Le mer. 29 mars 2023 à 09:54, Yufan Sheng <sy...@gmail.com> a écrit :
> > >
> > > Hi SiNan,
> > >
> > > In the flink world, we don't always rely on the schema information
> > > provided by Pulsar or other connector systems. Flink application has
> > > its own (de)serialization schema logic, which treats the messages only
> > > in a binary format like a byte array.
> > >
> > > In flink-connector-pulsar, we only use the schema when the users want
> > > to do some evolution check. Otherwise, we will only send messages in
> > > BYTES schema.
> > >
> > > On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <li...@gmail.com>
> wrote:
> > > >
> > > > Hi yufan.
> > > > Can you describe a bit the usage scenario of byte schema in
> > > > flink-connector-pulsa?
> > > >
> > > >
> > > > Thanks,
> > > > sinan
> > > >
> > > > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
> > > >
> > > > > As the flink-connector-pulsar developer, I don't want to disable
> the
> > > > > BYTES schema upload. In my opinion, using BYTES schema means the
> users
> > > > > want to bypass the schema check and handle the schema validation by
> > > > > themselves.
> > > > >
> > > > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <li...@gmail.com>
> wrote:
> > > > > >
> > > > > > Hi, everyone.
> > > > > > When a user uploads bytes schema. We can warn the user and skip
> uploading
> > > > > > bytes schema.
> > > > > > Also check to see if the topic has a schema other than bytes.
> > > > > > 1. If yes, warn the user that it is not necessary to upload
> bytes schema.
> > > > > > You can subscribe to a topic using bytes schema.
> > > > > > 2. If there is no schema, warn the user that the topic does not
> have a
> > > > > > schema. The default is bytes schema, and there is no need to
> upload it.
> > > > > > Rather than simply throwing an exception rejecting the upload
> bytes
> > > > > schema.
> > > > > >
> > > > > >
> > > > > > Thanks,
> > > > > > sinan
> > > > > >
> > > > > >
> > > > > > Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二
> 上午1:15写道：
> > > > > >
> > > > > > > This change broke the Flink SQL Pulsar connector:
> > > > > > > https://github.com/streamnative/flink/issues/270
> > > > > > > So I propose to revert it.
> > > > > > >
> > > > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <
> labuladong@foxmail.com> a
> > > > > écrit :
> > > > > > > >
> > > > > > > > Hi pulsar community,
> > > > > > > >
> > > > > > > >
> > > > > > > > I'd like to discuss the behavior of schema uploading, for
> more
> > > > > context
> > > > > > > see https://github.com/apache/pulsar/issues/18825
> > > > > > > >
> > > > > > > >
> > > > > > > > I think that forbidding users to upload `BYTES` schema is a
> > > > > recommended
> > > > > > > way to solve this issue. But this may change the existing
> behavior, so
> > > > > do
> > > > > > > you have any suggestion about this issue?
> > > > > > > >
> > > > > > > >
> > > > > > > > Thanks,
> > > > > > > > Donglai
> > > > > > >
> > > > >
>
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by Enrico Olivelli <eo...@gmail.com>.
Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet
<bo...@gmail.com> ha scritto:
>
> The change was not reverted for 3.0.0-RC so in the current state the
> Flink Pulsar catalog won't work with Pulsar 3.0.
> To give more details, Flink Pulsar stores Flink Table metadata in the
> metadata of BYTES schema topics (those topics don't have messages,
> only the schema metadata of these topics is used).
I think that we should not introduce this breaking change, it will be
a big pain for the users in the Flink ecosystem
Enrico
>
> Le mer. 29 mars 2023 à 09:54, Yufan Sheng <sy...@gmail.com> a écrit :
> >
> > Hi SiNan,
> >
> > In the flink world, we don't always rely on the schema information
> > provided by Pulsar or other connector systems. Flink application has
> > its own (de)serialization schema logic, which treats the messages only
> > in a binary format like a byte array.
> >
> > In flink-connector-pulsar, we only use the schema when the users want
> > to do some evolution check. Otherwise, we will only send messages in
> > BYTES schema.
> >
> > On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <li...@gmail.com> wrote:
> > >
> > > Hi yufan.
> > > Can you describe a bit the usage scenario of byte schema in
> > > flink-connector-pulsa?
> > >
> > >
> > > Thanks,
> > > sinan
> > >
> > > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
> > >
> > > > As the flink-connector-pulsar developer, I don't want to disable the
> > > > BYTES schema upload. In my opinion, using BYTES schema means the users
> > > > want to bypass the schema check and handle the schema validation by
> > > > themselves.
> > > >
> > > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <li...@gmail.com> wrote:
> > > > >
> > > > > Hi, everyone.
> > > > > When a user uploads bytes schema. We can warn the user and skip uploading
> > > > > bytes schema.
> > > > > Also check to see if the topic has a schema other than bytes.
> > > > > 1. If yes, warn the user that it is not necessary to upload bytes schema.
> > > > > You can subscribe to a topic using bytes schema.
> > > > > 2. If there is no schema, warn the user that the topic does not have a
> > > > > schema. The default is bytes schema, and there is no need to upload it.
> > > > > Rather than simply throwing an exception rejecting the upload bytes
> > > > schema.
> > > > >
> > > > >
> > > > > Thanks,
> > > > > sinan
> > > > >
> > > > >
> > > > > Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二 上午1:15写道：
> > > > >
> > > > > > This change broke the Flink SQL Pulsar connector:
> > > > > > https://github.com/streamnative/flink/issues/270
> > > > > > So I propose to revert it.
> > > > > >
> > > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <la...@foxmail.com> a
> > > > écrit :
> > > > > > >
> > > > > > > Hi pulsar community,
> > > > > > >
> > > > > > >
> > > > > > > I'd like to discuss the behavior of schema uploading, for more
> > > > context
> > > > > > see https://github.com/apache/pulsar/issues/18825
> > > > > > >
> > > > > > >
> > > > > > > I think that forbidding users to upload `BYTES` schema is a
> > > > recommended
> > > > > > way to solve this issue. But this may change the existing behavior, so
> > > > do
> > > > > > you have any suggestion about this issue?
> > > > > > >
> > > > > > >
> > > > > > > Thanks,
> > > > > > > Donglai
> > > > > >
> > > >
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by Christophe Bornet <bo...@gmail.com>.
The change was not reverted for 3.0.0-RC so in the current state the
Flink Pulsar catalog won't work with Pulsar 3.0.
To give more details, Flink Pulsar stores Flink Table metadata in the
metadata of BYTES schema topics (those topics don't have messages,
only the schema metadata of these topics is used).
Le mer. 29 mars 2023 à 09:54, Yufan Sheng <sy...@gmail.com> a écrit :
>
> Hi SiNan,
>
> In the flink world, we don't always rely on the schema information
> provided by Pulsar or other connector systems. Flink application has
> its own (de)serialization schema logic, which treats the messages only
> in a binary format like a byte array.
>
> In flink-connector-pulsar, we only use the schema when the users want
> to do some evolution check. Otherwise, we will only send messages in
> BYTES schema.
>
> On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <li...@gmail.com> wrote:
> >
> > Hi yufan.
> > Can you describe a bit the usage scenario of byte schema in
> > flink-connector-pulsa?
> >
> >
> > Thanks,
> > sinan
> >
> > Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
> >
> > > As the flink-connector-pulsar developer, I don't want to disable the
> > > BYTES schema upload. In my opinion, using BYTES schema means the users
> > > want to bypass the schema check and handle the schema validation by
> > > themselves.
> > >
> > > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <li...@gmail.com> wrote:
> > > >
> > > > Hi, everyone.
> > > > When a user uploads bytes schema. We can warn the user and skip uploading
> > > > bytes schema.
> > > > Also check to see if the topic has a schema other than bytes.
> > > > 1. If yes, warn the user that it is not necessary to upload bytes schema.
> > > > You can subscribe to a topic using bytes schema.
> > > > 2. If there is no schema, warn the user that the topic does not have a
> > > > schema. The default is bytes schema, and there is no need to upload it.
> > > > Rather than simply throwing an exception rejecting the upload bytes
> > > schema.
> > > >
> > > >
> > > > Thanks,
> > > > sinan
> > > >
> > > >
> > > > Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二 上午1:15写道：
> > > >
> > > > > This change broke the Flink SQL Pulsar connector:
> > > > > https://github.com/streamnative/flink/issues/270
> > > > > So I propose to revert it.
> > > > >
> > > > > Le ven. 9 déc. 2022 à 11:57, labuladong <la...@foxmail.com> a
> > > écrit :
> > > > > >
> > > > > > Hi pulsar community,
> > > > > >
> > > > > >
> > > > > > I'd like to discuss the behavior of schema uploading, for more
> > > context
> > > > > see https://github.com/apache/pulsar/issues/18825
> > > > > >
> > > > > >
> > > > > > I think that forbidding users to upload `BYTES` schema is a
> > > recommended
> > > > > way to solve this issue. But this may change the existing behavior, so
> > > do
> > > > > you have any suggestion about this issue?
> > > > > >
> > > > > >
> > > > > > Thanks,
> > > > > > Donglai
> > > > >
> > >
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by Yufan Sheng <sy...@gmail.com>.
Hi SiNan,
In the flink world, we don't always rely on the schema information
provided by Pulsar or other connector systems. Flink application has
its own (de)serialization schema logic, which treats the messages only
in a binary format like a byte array.
In flink-connector-pulsar, we only use the schema when the users want
to do some evolution check. Otherwise, we will only send messages in
BYTES schema.
On Tue, Mar 28, 2023 at 10:06 AM SiNan Liu <li...@gmail.com> wrote:
>
> Hi yufan.
> Can you describe a bit the usage scenario of byte schema in
> flink-connector-pulsa?
>
>
> Thanks,
> sinan
>
> Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
>
> > As the flink-connector-pulsar developer, I don't want to disable the
> > BYTES schema upload. In my opinion, using BYTES schema means the users
> > want to bypass the schema check and handle the schema validation by
> > themselves.
> >
> > On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <li...@gmail.com> wrote:
> > >
> > > Hi, everyone.
> > > When a user uploads bytes schema. We can warn the user and skip uploading
> > > bytes schema.
> > > Also check to see if the topic has a schema other than bytes.
> > > 1. If yes, warn the user that it is not necessary to upload bytes schema.
> > > You can subscribe to a topic using bytes schema.
> > > 2. If there is no schema, warn the user that the topic does not have a
> > > schema. The default is bytes schema, and there is no need to upload it.
> > > Rather than simply throwing an exception rejecting the upload bytes
> > schema.
> > >
> > >
> > > Thanks,
> > > sinan
> > >
> > >
> > > Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二 上午1:15写道：
> > >
> > > > This change broke the Flink SQL Pulsar connector:
> > > > https://github.com/streamnative/flink/issues/270
> > > > So I propose to revert it.
> > > >
> > > > Le ven. 9 déc. 2022 à 11:57, labuladong <la...@foxmail.com> a
> > écrit :
> > > > >
> > > > > Hi pulsar community,
> > > > >
> > > > >
> > > > > I'd like to discuss the behavior of schema uploading, for more
> > context
> > > > see https://github.com/apache/pulsar/issues/18825
> > > > >
> > > > >
> > > > > I think that forbidding users to upload `BYTES` schema is a
> > recommended
> > > > way to solve this issue. But this may change the existing behavior, so
> > do
> > > > you have any suggestion about this issue?
> > > > >
> > > > >
> > > > > Thanks,
> > > > > Donglai
> > > >
> >
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by SiNan Liu <li...@gmail.com>.
Hi yufan.
Can you describe a bit the usage scenario of byte schema in
flink-connector-pulsa?
Thanks,
sinan
Yufan Sheng <sy...@gmail.com> 于 2023年3月28日周二 上午9:53写道：
> As the flink-connector-pulsar developer, I don't want to disable the
> BYTES schema upload. In my opinion, using BYTES schema means the users
> want to bypass the schema check and handle the schema validation by
> themselves.
>
> On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <li...@gmail.com> wrote:
> >
> > Hi, everyone.
> > When a user uploads bytes schema. We can warn the user and skip uploading
> > bytes schema.
> > Also check to see if the topic has a schema other than bytes.
> > 1. If yes, warn the user that it is not necessary to upload bytes schema.
> > You can subscribe to a topic using bytes schema.
> > 2. If there is no schema, warn the user that the topic does not have a
> > schema. The default is bytes schema, and there is no need to upload it.
> > Rather than simply throwing an exception rejecting the upload bytes
> schema.
> >
> >
> > Thanks,
> > sinan
> >
> >
> > Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二 上午1:15写道：
> >
> > > This change broke the Flink SQL Pulsar connector:
> > > https://github.com/streamnative/flink/issues/270
> > > So I propose to revert it.
> > >
> > > Le ven. 9 déc. 2022 à 11:57, labuladong <la...@foxmail.com> a
> écrit :
> > > >
> > > > Hi pulsar community,
> > > >
> > > >
> > > > I'd like to discuss the behavior of schema uploading, for more
> context
> > > see https://github.com/apache/pulsar/issues/18825
> > > >
> > > >
> > > > I think that forbidding users to upload `BYTES` schema is a
> recommended
> > > way to solve this issue. But this may change the existing behavior, so
> do
> > > you have any suggestion about this issue?
> > > >
> > > >
> > > > Thanks,
> > > > Donglai
> > >
>
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by Yufan Sheng <sy...@gmail.com>.
As the flink-connector-pulsar developer, I don't want to disable the
BYTES schema upload. In my opinion, using BYTES schema means the users
want to bypass the schema check and handle the schema validation by
themselves.
On Tue, Mar 28, 2023 at 8:58 AM SiNan Liu <li...@gmail.com> wrote:
>
> Hi, everyone.
> When a user uploads bytes schema. We can warn the user and skip uploading
> bytes schema.
> Also check to see if the topic has a schema other than bytes.
> 1. If yes, warn the user that it is not necessary to upload bytes schema.
> You can subscribe to a topic using bytes schema.
> 2. If there is no schema, warn the user that the topic does not have a
> schema. The default is bytes schema, and there is no need to upload it.
> Rather than simply throwing an exception rejecting the upload bytes schema.
>
>
> Thanks,
> sinan
>
>
> Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二 上午1:15写道：
>
> > This change broke the Flink SQL Pulsar connector:
> > https://github.com/streamnative/flink/issues/270
> > So I propose to revert it.
> >
> > Le ven. 9 déc. 2022 à 11:57, labuladong <la...@foxmail.com> a écrit :
> > >
> > > Hi pulsar community,
> > >
> > >
> > > I'd like to discuss the behavior of schema uploading, for more context
> > see https://github.com/apache/pulsar/issues/18825
> > >
> > >
> > > I think that forbidding users to upload `BYTES` schema is a recommended
> > way to solve this issue. But this may change the existing behavior, so do
> > you have any suggestion about this issue?
> > >
> > >
> > > Thanks,
> > > Donglai
> >
Re: [DISCUSS] forbid user to upload `BYTES` schema
Posted by SiNan Liu <li...@gmail.com>.
Hi, everyone.
When a user uploads bytes schema. We can warn the user and skip uploading
bytes schema.
Also check to see if the topic has a schema other than bytes.
1. If yes, warn the user that it is not necessary to upload bytes schema.
You can subscribe to a topic using bytes schema.
2. If there is no schema, warn the user that the topic does not have a
schema. The default is bytes schema, and there is no need to upload it.
Rather than simply throwing an exception rejecting the upload bytes schema.
Thanks,
sinan
Christophe Bornet <bo...@gmail.com> 于 2023年3月28日周二 上午1:15写道：
> This change broke the Flink SQL Pulsar connector:
> https://github.com/streamnative/flink/issues/270
> So I propose to revert it.
>
> Le ven. 9 déc. 2022 à 11:57, labuladong <la...@foxmail.com> a écrit :
> >
> > Hi pulsar community,
> >
> >
> > I'd like to discuss the behavior of schema uploading, for more context
> see https://github.com/apache/pulsar/issues/18825
> >
> >
> > I think that forbidding users to upload `BYTES` schema is a recommended
> way to solve this issue. But this may change the existing behavior, so do
> you have any suggestion about this issue?
> >
> >
> > Thanks,
> > Donglai
>