From 7b4604e35cc9ab37907db06e6ee8153c90a9e42c Mon Sep 17 00:00:00 2001 From: labuladong Date: Mon, 21 Nov 2022 14:41:01 +0800 Subject: [PATCH 1/2] improve doc --- site2/docs/schema-get-started.md | 40 +++++++++++++++---------------- site2/docs/schema-overview.md | 41 +++++++++++++++++++++++--------- site2/docs/schema-understand.md | 19 ++++++++------- 3 files changed, 60 insertions(+), 40 deletions(-) diff --git a/site2/docs/schema-get-started.md b/site2/docs/schema-get-started.md index 5b6cea677e3b4..f01156b96639f 100644 --- a/site2/docs/schema-get-started.md +++ b/site2/docs/schema-get-started.md @@ -28,7 +28,7 @@ This example demonstrates how to construct a [string schema](schema-understand.m ```java Consumer consumer = client.newConsumer(Schema.STRING).subscribe(); - consumer.receive(); + Message message = consumer.receive(); ``` ## Construct a key/value schema @@ -37,29 +37,29 @@ This example shows how to construct a [key/value schema](schema-understand.md#ke 1. Construct a key/value schema with `INLINE` encoding type. - ```java - Schema> kvSchema = Schema.KeyValue( - Schema.INT32, - Schema.STRING, - KeyValueEncodingType.INLINE - ); - ``` + ```java + Schema> kvSchema = Schema.KeyValue( + Schema.INT32, + Schema.STRING, + KeyValueEncodingType.INLINE + ); + ``` 2. Optionally, construct a key/value schema with `SEPARATED` encoding type. - ```java - Schema> kvSchema = Schema.KeyValue( - Schema.INT32, - Schema.STRING, - KeyValueEncodingType.SEPARATED - ); - ``` + ```java + Schema> kvSchema = Schema.KeyValue( + Schema.INT32, + Schema.STRING, + KeyValueEncodingType.SEPARATED + ); + ``` 3. Produce messages using a key/value schema. ```java Producer> producer = client.newProducer(kvSchema) - .topic(TOPIC) + .topic(topicName) .create(); final int key = 100; @@ -67,8 +67,8 @@ This example shows how to construct a [key/value schema](schema-understand.md#ke // send the key/value message producer.newMessage() - .value(new KeyValue(key, value)) - .send(); + .value(new KeyValue(key, value)) + .send(); ``` 4. Consume messages using a key/value schema. @@ -76,8 +76,8 @@ This example shows how to construct a [key/value schema](schema-understand.md#ke ```java Consumer> consumer = client.newConsumer(kvSchema) ... - .topic(TOPIC) - .subscriptionName(SubscriptionName).subscribe(); + .topic(topicName) + .subscriptionName(subscriptionName).subscribe(); // receive key/value pair Message> msg = consumer.receive(); diff --git a/site2/docs/schema-overview.md b/site2/docs/schema-overview.md index 9560a95443f63..2e09576ce5c80 100644 --- a/site2/docs/schema-overview.md +++ b/site2/docs/schema-overview.md @@ -60,17 +60,17 @@ This diagram illustrates how schema works on the Producer side. 5. If no, the broker verifies whether a schema can be automatically created in this namespace: - * If `isAllowAutoUpdateSchema` sets to **true**, then a schema can be created, and the broker validates the schema based on the schema compatibility check strategy defined for the topic. + * If `isAllowAutoUpdateSchemaEnabled` sets to **true**, then a schema can be created, and the broker validates the schema based on the schema compatibility check strategy defined for the topic. - * If `isAllowAutoUpdateSchema` sets to **false**, then a schema can not be created, and the producer is rejected to connect to the broker. + * If `isAllowAutoUpdateSchemaEnabled` sets to **false**, then a schema can not be created, and the producer is rejected to connect to the broker. :::tip - `isAllowAutoUpdateSchema` can be set via **Pulsar admin API** or **REST API.** + `isAllowAutoUpdateSchemaEnabled` can be set via **Pulsar admin API** or **REST API.** - For how to set `isAllowAutoUpdateSchema` via Pulsar admin API, see [Manage AutoUpdate Strategy](admin-api-schemas.md#manage-autoupdate-strategy). + For how to set `isAllowAutoUpdateSchemaEnabled` via Pulsar admin API, see [Manage AutoUpdate Strategy](admin-api-schemas.md#manage-autoupdate-strategy). - ::: + ::: 6. If the schema is allowed to be updated, then the compatible strategy check is performed. @@ -94,9 +94,9 @@ This diagram illustrates how schema works on the consumer side. 4. If a topic does not have all of them (a schema/data/a local consumer and a local producer): - * If `isAllowAutoUpdateSchema` sets to **true**, then the consumer registers a schema and it is connected to a broker. + * If `isAllowAutoUpdateSchemaEnabled` sets to **true**, then the consumer registers a schema and it is connected to a broker. - * If `isAllowAutoUpdateSchema` sets to **false**, then the consumer is rejected to connect to a broker. + * If `isAllowAutoUpdateSchemaEnabled` sets to **false**, then the consumer is rejected to connect to a broker. 5. If a topic has one of them (a schema/data/a local consumer and a local producer), then the schema compatibility check is performed. @@ -112,12 +112,19 @@ This diagram illustrates how schema works on the consumer side. You can use language-specific types of data when constructing and handling messages from simple data types like `string` to more complex application-specific types. -For example, you are using the _User_ class to define the messages sent to Pulsar topics. +For example, you are using the `User` class to define the messages sent to Pulsar topics. ```java public class User { - String name; - int age; + public String name; + public int age; + + User() {} + + User(String name, int age) { + this.name = name; + this.age = age; + } } ``` @@ -136,14 +143,26 @@ producer.send(message); **With a schema** -This example constructs a producer with the _JSONSchema_, and you can send the _User_ class to topics directly without worrying about how to serialize POJOs into bytes. +This example constructs a producer with the `JSONSchema`, and you can send the `User` class to topics directly without worrying about how to serialize POJOs into bytes. ```java +// send with json schema Producer producer = client.newProducer(JSONSchema.of(User.class)) .topic(topic) .create(); User user = new User("Tom", 28); producer.send(user); + +// receive with json schema +Consumer consumer = client.newConsumer(JSONSchema.of(User.class)) + .topic(schemaTopic) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscriptionName("schema-sub") + .subscribe(); + +Message message = consumer.receive(); +User user = message.getValue(); +assert user.age == 28 && user.name.equals("Tom"); ``` ## What's next? diff --git a/site2/docs/schema-understand.md b/site2/docs/schema-understand.md index 64cb6c9b6d8ea..10471c6f75176 100644 --- a/site2/docs/schema-understand.md +++ b/site2/docs/schema-understand.md @@ -87,7 +87,7 @@ Currently, Pulsar supports the following complex types: You can choose the encoding type when constructing the key/value schema.： * `INLINE` - Key/value pairs are encoded together in the message payload. -* `SEPARATED` - see [Construct a key/value schema](schema-get-started.md#construct-a-keyvalue-schema). +* `SEPARATED` - Key is stored as message key, while value is stored as message payload. #### `Struct` schema @@ -148,10 +148,11 @@ Producer producer = client.newProducer(JSONSchema.of(SensorReadin The table below lists the possible scenarios when this connection attempt occurs and what happens in each scenario: -| Scenario | What happens | -| --- | --- | -|
  • No schema exists for the topic.
    • | (1) The producer is created using the given schema. (2) Since no existing schema is compatible with the `SensorReading` schema, the schema is transmitted to the broker and stored. (3) Any consumer created using the same schema or topic can consume messages from the `sensor-data` topic. | -|
  • A schema already exists.
  • The producer connects using the same schema that is already stored.
    • | (1) The schema is transmitted to the broker. (2) The broker determines that the schema is compatible. (3) The broker attempts to store the schema in [BookKeeper](concepts-architecture-overview.md#persistent-storage) but then determines that it's already stored, so it is used to tag produced messages. |
  • A schema already exists.
  • The producer connects using a new schema that is compatible.
    • | (1) The schema is transmitted to the broker. (2) The broker determines that the schema is compatible and stores the new schema as the current version (with a new version number). | +| Scenario | What happens | +|-----------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +|
  • No schema exists for the topic.
    • | (1) The producer is created using the given schema.
    (2) Since no existing schema is compatible with the `SensorReading` schema, the schema is transmitted to the broker and stored.
    (3) Any consumer created using the same schema or topic can consume messages from the `sensor-data` topic. | +|
  • A schema already exists.
  • The producer connects using the same schema that is already stored.
    • | (1) The schema is transmitted to the broker.
    (2) The broker determines that the schema is compatible.
    (3) The broker attempts to store the schema in [BookKeeper](concepts-architecture-overview.md#persistent-storage) but then determines that it's already stored, so it is used to tag produced messages. | +|
  • A schema already exists.
  • The producer connects using a new schema that is compatible.
    • | (1) The schema is transmitted to the broker.
    (2) The broker determines that the schema is compatible and stores the new schema as the current version (with a new version number). | ## Schema AutoUpdate @@ -179,9 +180,9 @@ For a producer, the `AutoUpdate` happens in the following cases: * If the schema is not registered: - * If `isAllowAutoUpdateSchema` sets to **false**, the producer is rejected to connect to a broker. + * If `isAllowAutoUpdateSchemaEnabled` sets to **false**, the producer is rejected to connect to a broker. - * If `isAllowAutoUpdateSchema` sets to **true**: + * If `isAllowAutoUpdateSchemaEnabled` sets to **true**: * If the schema passes the compatibility check, then the broker registers a new schema automatically for the topic and the producer is connected. @@ -199,9 +200,9 @@ For a consumer, the `AutoUpdate` happens in the following cases: * If a topic does not have all of them (a schema/data/a local consumer and a local producer): - * If `isAllowAutoUpdateSchema` sets to **true**, then the consumer registers a schema and it is connected to a broker. + * If `isAllowAutoUpdateSchemaEnabled` sets to **true**, then the consumer registers a schema and it is connected to a broker. - * If `isAllowAutoUpdateSchema` sets to **false**, then the consumer is rejected to connect to a broker. + * If `isAllowAutoUpdateSchemaEnabled` sets to **false**, then the consumer is rejected to connect to a broker. * If a topic has one of them (a schema/data/a local consumer and a local producer), then the schema compatibility check is performed. From a246cb8805df2e1246a37dc489c0898cb8ee2eae Mon Sep 17 00:00:00 2001 From: labuladong Date: Mon, 21 Nov 2022 17:03:17 +0800 Subject: [PATCH 2/2] improve avro schema --- site2/docs/schema-get-started.md | 20 ++++++++++++++++---- site2/docs/schema-overview.md | 12 ++++++------ site2/docs/schema-understand.md | 26 ++++++++++++-------------- 3 files changed, 34 insertions(+), 24 deletions(-) diff --git a/site2/docs/schema-get-started.md b/site2/docs/schema-get-started.md index f01156b96639f..d3fad3c45de16 100644 --- a/site2/docs/schema-get-started.md +++ b/site2/docs/schema-get-started.md @@ -142,16 +142,28 @@ You can define the `struct` schema using the `GenericSchemaBuilder`, generate a RecordSchemaBuilder recordSchemaBuilder = SchemaBuilder.record("schemaName"); recordSchemaBuilder.field("intField").type(SchemaType.INT32); SchemaInfo schemaInfo = recordSchemaBuilder.build(SchemaType.AVRO); + + Consumer consumer = client.newConsumer(Schema.generic(schemaInfo)) + .topic(topicName) + .subscriptionName(subscriptionName) + .subscribe(); - Producer producer = client.newProducer(Schema.generic(schemaInfo)).create(); + Producer producer = client.newProducer(Schema.generic(schemaInfo)) + .topic(topicName) + .create(); ``` 2. Use `RecordBuilder` to build the struct records. ```java - producer.newMessage().value(schema.newRecordBuilder() - .set("intField", 32) - .build()).send(); + GenericSchemaImpl schema = GenericAvroSchema.of(schemaInfo); + // send message + GenericRecord record = schema.newRecordBuilder().set("intField", 32).build(); + producer.newMessage().value(record).send(); + // receive message + Message msg = consumer.receive(); + + Assert.assertEquals(msg.getValue().getField("intField"), 32); ``` diff --git a/site2/docs/schema-overview.md b/site2/docs/schema-overview.md index 2e09576ce5c80..6da46e17924c5 100644 --- a/site2/docs/schema-overview.md +++ b/site2/docs/schema-overview.md @@ -60,15 +60,15 @@ This diagram illustrates how schema works on the Producer side. 5. If no, the broker verifies whether a schema can be automatically created in this namespace: - * If `isAllowAutoUpdateSchemaEnabled` sets to **true**, then a schema can be created, and the broker validates the schema based on the schema compatibility check strategy defined for the topic. + * If `isAllowAutoUpdateSchema` sets to **true**, then a schema can be created, and the broker validates the schema based on the schema compatibility check strategy defined for the topic. - * If `isAllowAutoUpdateSchemaEnabled` sets to **false**, then a schema can not be created, and the producer is rejected to connect to the broker. + * If `isAllowAutoUpdateSchema` sets to **false**, then a schema can not be created, and the producer is rejected to connect to the broker. :::tip - `isAllowAutoUpdateSchemaEnabled` can be set via **Pulsar admin API** or **REST API.** + `isAllowAutoUpdateSchema` can be set in broker level via config file `conf/standalone.conf` or `conf/broker.conf`. It can also be set in namespace level via **Pulsar admin API** or **REST API**. - For how to set `isAllowAutoUpdateSchemaEnabled` via Pulsar admin API, see [Manage AutoUpdate Strategy](admin-api-schemas.md#manage-autoupdate-strategy). + For how to set `isAllowAutoUpdateSchema` via Pulsar admin API, see [Manage AutoUpdate Strategy](admin-api-schemas.md#manage-autoupdate-strategy). ::: @@ -94,9 +94,9 @@ This diagram illustrates how schema works on the consumer side. 4. If a topic does not have all of them (a schema/data/a local consumer and a local producer): - * If `isAllowAutoUpdateSchemaEnabled` sets to **true**, then the consumer registers a schema and it is connected to a broker. + * If `isAllowAutoUpdateSchema` sets to **true**, then the consumer registers a schema and it is connected to a broker. - * If `isAllowAutoUpdateSchemaEnabled` sets to **false**, then the consumer is rejected to connect to a broker. + * If `isAllowAutoUpdateSchema` sets to **false**, then the consumer is rejected to connect to a broker. 5. If a topic has one of them (a schema/data/a local consumer and a local producer), then the schema compatibility check is performed. diff --git a/site2/docs/schema-understand.md b/site2/docs/schema-understand.md index 10471c6f75176..7945d2f71f0f6 100644 --- a/site2/docs/schema-understand.md +++ b/site2/docs/schema-understand.md @@ -168,25 +168,23 @@ For a producer, the `AutoUpdate` happens in the following cases: * If a **producer doesn’t carry a schema**: - * If `isSchemaValidationEnforced` or `schemaValidationEnforced` is **disabled** in the namespace to which the topic belongs, the producer is allowed to connect to the topic and produce data. + * If `isSchemaValidationEnforced` or `schemaValidationEnforced` is **disabled** in the namespace to which the topic belongs, the producer is allowed to connect to the topic and produce data. - * If `isSchemaValidationEnforced` or `schemaValidationEnforced` is **enabled** in the namespace to which the topic belongs, the producer is rejected and disconnected. + * If `isSchemaValidationEnforced` or `schemaValidationEnforced` is **enabled** in the namespace to which the topic belongs, the producer is rejected and disconnected. - * If a **producer carries a schema**: + * If a **producer carries a schema**, the broker will perform the compatibility check based on the configured compatibility check strategy of the namespace to which the topic belongs: - A broker performs the compatibility check based on the configured compatibility check strategy of the namespace to which the topic belongs. + * If the schema is registered, the producer is connected to a broker. - * If the schema is registered, a producer is connected to a broker. + * If the schema is not registered: - * If the schema is not registered: + * If `isAllowAutoUpdateSchema` sets to **false**, the producer is rejected to connect to the broker. - * If `isAllowAutoUpdateSchemaEnabled` sets to **false**, the producer is rejected to connect to a broker. - - * If `isAllowAutoUpdateSchemaEnabled` sets to **true**: + * If `isAllowAutoUpdateSchema` sets to **true**: - * If the schema passes the compatibility check, then the broker registers a new schema automatically for the topic and the producer is connected. + * If the schema passes the compatibility check, then the broker registers a new schema automatically for the topic and the producer is connected. - * If the schema does not pass the compatibility check, then the broker does not register a schema and the producer is rejected to connect to a broker. + * If the schema does not pass the compatibility check, then the broker does not register a schema and the producer is rejected to connect to a broker. ![AutoUpdate Producer](/assets/schema-producer.png) @@ -200,11 +198,11 @@ For a consumer, the `AutoUpdate` happens in the following cases: * If a topic does not have all of them (a schema/data/a local consumer and a local producer): - * If `isAllowAutoUpdateSchemaEnabled` sets to **true**, then the consumer registers a schema and it is connected to a broker. + * If `isAllowAutoUpdateSchema` sets to **true**, then the consumer registers the schema and is connected to the broker. - * If `isAllowAutoUpdateSchemaEnabled` sets to **false**, then the consumer is rejected to connect to a broker. + * If `isAllowAutoUpdateSchema` sets to **false**, then the consumer is rejected to connect to the broker. - * If a topic has one of them (a schema/data/a local consumer and a local producer), then the schema compatibility check is performed. + * If a topic has one of them (a schema/data/a local consumer or a local producer), then the schema compatibility check is performed. * If the schema passes the compatibility check, then the consumer is connected to the broker.