Skip to content

[improve][doc] Pulsar Schema #18556

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
labuladong wants to merge 2 commits into from
Closed

[improve][doc] Pulsar Schema #18556

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7b4604e
improve doc
labuladong Nov 21, 2022
a246cb8
improve avro schema
labuladong Nov 21, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 36 additions & 24 deletions site2/docs/schema-get-started.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ This example demonstrates how to construct a [string schema](schema-understand.m

```java
Consumer<String> consumer = client.newConsumer(Schema.STRING).subscribe();
consumer.receive();
Message<String> message = consumer.receive();
```

## Construct a key/value schema
Expand All @@ -37,47 +37,47 @@ 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<KeyValue<Integer, String>> kvSchema = Schema.KeyValue(
Schema.INT32,
Schema.STRING,
KeyValueEncodingType.INLINE
);
```
```java
Schema<KeyValue<Integer, String>> kvSchema = Schema.KeyValue(
Schema.INT32,
Schema.STRING,
KeyValueEncodingType.INLINE
);
```

2. Optionally, construct a key/value schema with `SEPARATED` encoding type.

```java
Schema<KeyValue<Integer, String>> kvSchema = Schema.KeyValue(
Schema.INT32,
Schema.STRING,
KeyValueEncodingType.SEPARATED
);
```
```java
Schema<KeyValue<Integer, String>> kvSchema = Schema.KeyValue(
Schema.INT32,
Schema.STRING,
KeyValueEncodingType.SEPARATED
);
```

3. Produce messages using a key/value schema.

```java
Producer<KeyValue<Integer, String>> producer = client.newProducer(kvSchema)
.topic(TOPIC)
.topic(topicName)
.create();

final int key = 100;
final String value = "value-100";

// 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.

```java
Consumer<KeyValue<Integer, String>> consumer = client.newConsumer(kvSchema)
...
.topic(TOPIC)
.subscriptionName(SubscriptionName).subscribe();
.topic(topicName)
.subscriptionName(subscriptionName).subscribe();

// receive key/value pair
Message<KeyValue<Integer, String>> msg = consumer.receive();
Expand Down Expand Up @@ -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<GenericRecord> consumer = client.newConsumer(Schema.generic(schemaInfo))
.topic(topicName)
.subscriptionName(subscriptionName)
.subscribe();

Producer<GenericRecord> producer = client.newProducer(Schema.generic(schemaInfo)).create();
Producer<GenericRecord> 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<GenericRecord> msg = consumer.receive();

Assert.assertEquals(msg.getValue().getField("intField"), 32);
```

</TabItem>
Expand Down
31 changes: 25 additions & 6 deletions site2/docs/schema-overview.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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;
}
}
```

Expand All @@ -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<User> producer = client.newProducer(JSONSchema.of(User.class))
.topic(topic)
.create();
User user = new User("Tom", 28);
producer.send(user);

// receive with json schema
Consumer<User> consumer = client.newConsumer(JSONSchema.of(User.class))
.topic(schemaTopic)
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
.subscriptionName("schema-sub")
.subscribe();

Message<User> message = consumer.receive();
User user = message.getValue();
assert user.age == 28 && user.name.equals("Tom");
```

## What's next?
Expand Down
37 changes: 18 additions & 19 deletions site2/docs/schema-understand.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -148,10 +148,11 @@ Producer<SensorReading> 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 |
| --- | --- |
| <li>No schema exists for the topic. </li> | (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. |
| <li>A schema already exists. </li><li>The producer connects using the same schema that is already stored. </li> | (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. | <li>A schema already exists. </li><li>The producer connects using a new schema that is compatible. </li> | (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 |
|-----------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| <li>No schema exists for the topic. </li> | (1) The producer is created using the given schema. <br/>(2) Since no existing schema is compatible with the `SensorReading` schema, the schema is transmitted to the broker and stored. <br/>(3) Any consumer created using the same schema or topic can consume messages from the `sensor-data` topic. |
| <li>A schema already exists. </li><li>The producer connects using the same schema that is already stored. </li> | (1) The schema is transmitted to the broker. <br/>(2) The broker determines that the schema is compatible. <br/>(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. |
| <li>A schema already exists. </li><li>The producer connects using a new schema that is compatible. </li> | (1) The schema is transmitted to the broker. <br/>(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

Expand All @@ -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)

Expand All @@ -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.

Expand Down