> 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 <syh...@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 <peng...@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 <peng...@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 < > mmarsh...@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 <liusinan1...@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 <eolive...@gmail.com> 于2023年4月14日周五 22:50写道： > > >> > > > >> > > Il giorno ven 14 apr 2023 alle ore 16:48 Christophe Bornet > > >> > > <bornet.ch...@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 <syh...@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 < > > >> liusinan1...@gmail.com> > > >> > > wrote: > > >> > > > > > > > >> > > > > > Hi yufan. > > >> > > > > > Can you describe a bit the usage scenario of byte schema in > > >> > > > > > flink-connector-pulsa? > > >> > > > > > > > >> > > > > > > > >> > > > > > Thanks, > > >> > > > > > sinan > > >> > > > > > > > >> > > > > > Yufan Sheng <syh...@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 < > > >> liusinan1...@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 <bornet.ch...@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 < > > >> > > labulad...@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 > > >> > > > > > > > > > > >> > > > > > > > > >> > > > > >> > > > >
