diff --git a/site2/docs/schema-get-started.md b/site2/docs/schema-get-started.md index 5b6cea677e3b4..d3fad3c45de16 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(); @@ -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 9560a95443f63..6da46e17924c5 100644 --- a/site2/docs/schema-overview.md +++ b/site2/docs/schema-overview.md @@ -66,11 +66,11 @@ This diagram illustrates how schema works on the Producer side. :::tip - `isAllowAutoUpdateSchema` 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 `isAllowAutoUpdateSchema` 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. @@ -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..7945d2f71f0f6 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 @@ -167,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**: - - A broker performs the compatibility check based on the configured compatibility check strategy of the namespace to which the topic belongs. + * 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: - * If the schema is registered, a producer is connected to a broker. + * If the schema is registered, the 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 a broker. + * If `isAllowAutoUpdateSchema` sets to **false**, the producer is rejected to connect to the broker. - * If `isAllowAutoUpdateSchema` 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) @@ -199,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 `isAllowAutoUpdateSchema` 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 `isAllowAutoUpdateSchema` 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.