apache / apache/pulsar

PIP-236: Record schema in the request and carry to the broker when subscribing with AUTO_CONSUME schema.

Open
#19,113 1 comment 0 reactions 0 assignees View on GitHub
Stale type/PIP
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

### Motivation

Fixed the failure to use schema to create consumer after using AUTO-CONSUME consumer to subscribe an empty topic, and Broker returned the error message as IncompatibleSchemaException("Topic does not have schema to check").
https://github.com/apache/pulsar/blob/ed33fb399e661e4d47baeaaa8d0cdb3bfadc9546/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1037
https://github.com/apache/pulsar/blob/ed33fb399e661e4d47baeaaa8d0cdb3bfadc9546/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1147-L1152
https://github.com/apache/pulsar/blob/ed33fb399e661e4d47baeaaa8d0cdb3bfadc9546/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L3054-L3071
https://github.com/apache/pulsar/blob/ed33fb399e661e4d47baeaaa8d0cdb3bfadc9546/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java#L1162-L1177
We should record whether the active consumers of the Topic have one or more consumers whose SchemaType is not AUTO_CONSUME.

### Goal

1. On the client side, we add AutoConsume enum in the Schema.Type, and add protocol version in the ProtocolVersion.
2. On the broker side, the SchemaType containing AUTO_CONSUME is recorded in `org.apache.pulsar.broker.service.Consumer`.

### API Changes

Protocal change: Schema.Type and ProtocolVersion

```protobuf
message Schema {
enum Type {
AutoConsume = 21;
}
}

enum ProtocolVersion {
v21 = 21; // Carry the AUTO_CONSUME schema to the Broker after this version
}
```

Record SchemaType in Consumer

```java
@Getter
private final SchemaType schemaType;
```

### Implementation

On the client side

1. Set "default" schemaInfo(schemaType=SchemaType.AUTO_CONSUME) in `org.apache.pulsar.client.impl.schema.AutoConsumeSchema`.
2. Get and set SchemaInfo with AUTO_CONSUME schema in `org.apache.pulsar.common.protocol.Commands#newSubscribe` when schemaType is AUTO_CONSUME and proto version is greater than or equal to `v21`.

On the broker side

1. When the schema is not null and schemaType not AUTO_CONSUME, then addSchemaIfIdleOrCheckCompatible.
2. Get the schemaType and record it in Consumer.
3. The active consumers schema of the Topic have one or more consumers schema whose not AUTO_CONSUME. Then `checkSchemaCompatibleForConsumer`.

```java
@Override
public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema) {
return hasSchema().thenCompose((hasSchema) -> {
int numActiveConsumersWithoutAutoSchema = subscriptions.values().stream()
.mapToInt(subscription -> subscription.getConsumers().stream()
.filter(consumer -> consumer.getSchemaType() != SchemaType.AUTO_CONSUME)
.toList().size())
.sum();
if (hasSchema
|| (!producers.isEmpty())
|| (numActiveConsumersWithoutAutoSchema != 0)
|| (ledger.getTotalSize() != 0)) {
return checkSchemaCompatibleForConsumer(schema);
} else {
return addSchema(schema).thenCompose(schemaVersion ->
CompletableFuture.completedFuture(null));
}
});
}
```

### Alternatives

1. On the client side, we add an optional field in the CommandSubscribe.

add optional bool check_schema_compatibility = 20 [default = true];

```protobuf
message CommandSubscribe {
optional bool check_schema_compatibility = 20 [default = true];
}
```

2. On the broker side, record checkSchemaCompatibility in `org.apache.pulsar.broker.service.Consumer`.

3. If checkSchemaCompatibility is true, schema compatibility check is required. If schemaType is AUTO_CONSUME, the checkSchemaCompatibility is false.

### Anything else?

No response

### Links
Discussion: https://lists.apache.org/thread/v7p88h7grqnbzocw34g6jvxjfw962kfd
Vote: https://lists.apache.org/thread/pvcdlbflofoj41ryo1lrn0zlhj15bwpv
PR: #17449

Contributor guide

Open the contributing guide

Research direction

Review PR #17449 and the referenced client and broker entry points first, including AutoConsumeSchema, Commands#newSubscribe, Consumer, ServerCnx, PersistentTopic, and NonPersistentTopic. Verify that AUTO_CONSUME is carried through protocol version v21, recorded for consumers, and that compatibility checks follow the stated active-consumer behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend-api-design, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Clearly specified
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.