-
Notifications
You must be signed in to change notification settings - Fork 61
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add tombstone support to listeners (#507)
* Add tombstone support for @PulsarListener * Add tombstone support for @PulsarReader * Add tombstone support for @ReactivePulsarListener Resolves #506
- Loading branch information
Showing
23 changed files
with
1,377 additions
and
161 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
65 changes: 65 additions & 0 deletions
65
...lsar-docs/src/main/antora/modules/ROOT/pages/reference/tombstones-reactive.adoc
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| [[tombstones-reactive]] | ||
| = Null Payloads and Log Compaction of 'Tombstone' Records | ||
|
|
||
| When using log compaction, you can send and receive messages with `null` payloads to identify the deletion of a key. | ||
| You can also receive `null` values for other reasons, such as a deserializer that might return `null` when it cannot deserialize a value. | ||
|
|
||
| [[tombstones-reactive.produce]] | ||
| == Producing Null Payloads | ||
| You can send a `null` value with the `ReactivePulsarTemplate` by passing a `null` message parameter value to one of the `send` methods, for example: | ||
| [source, java] | ||
| ---- | ||
| reactiveTemplate | ||
| .send(null, Schema.STRING) | ||
| .subscribe(); | ||
| ---- | ||
| NOTE: When sending null values you must specify the schema type as the system can not determine the type of the message from a `null` payload. | ||
|
|
||
| [[tombstones-reactive.consume]] | ||
| == Consuming Null Payloads | ||
| For `@ReactivePularListener`, the `null` payload is passed into the listener method based on the type of its message parameter as follows: | ||
| |=== | ||
| | Parameter type | Passed-in value | ||
|
|
||
| | primitive | ||
| | `null` | ||
|
|
||
| | user-defined | ||
| | `null` | ||
|
|
||
| | `org.apache.pulsar.client.api.Message<T>` | ||
| | non-null Pulsar message whose `getValue()` returns `null` | ||
|
|
||
| | `org.springframework.messaging.Message<T>` | ||
| | non-null Spring message whose `getPayload()` returns `PulsarNull` | ||
|
|
||
| | `Flux<org.apache.pulsar.client.api.Message<T>>` | ||
| | non-null flux whose entries are non-null Pulsar messages whose `getValue()` returns `null` | ||
|
|
||
| | `Flux<org.apache.pulsar.client.api.Messages<T>>` | ||
| | non-null flux whose entries are non-null Spring messages whose `getPayload()` returns `PulsarNull` | ||
|
|
||
| |=== | ||
|
|
||
| IMPORTANT: When the passed-in value is `null` (ie. single record listeners with primitive or user-defined types) you must use the `@Payload` parameter annotation with `required = false`. | ||
|
|
||
| IMPORTANT: When using the Spring `org.springframework.messaging.Message` for your listener payload type, its generic type information must be wide enough to accept `Message<PulsarNull>` (eg. `Message`, `Message<?>`, or `Message<Object>`). | ||
| This is due to the fact that the Spring Message does not allow null values for its payload and instead uses the `PulsarNull` placeholder. | ||
|
|
||
| If it is a tombstone message for a compacted log, you usually also need the key so that your application can determine which key was +++"+++`deleted`+++"+++. | ||
| The following example shows such a configuration: | ||
|
|
||
| [source, java] | ||
| ---- | ||
| @ReactivePulsarListener( | ||
| topics = "my-topic", | ||
| subscriptionName = "my-topic-sub", | ||
| schemaType = SchemaType.STRING) | ||
| Mono<Void> myListener( | ||
| @Payload(required = false) String msg, | ||
| @Header(PulsarHeaders.KEY) String key) { | ||
| ... | ||
| } | ||
| ---- | ||
|
|
||
| NOTE: When using a streaming message listener (`Flux`) the xref:reference/reactive-pulsar.adoc#reactive-pulsar-headers.streaming[header support is limited], so it less useful in the log compaction scenario. |
68 changes: 68 additions & 0 deletions
68
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/tombstones.adoc
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| [[tombstones]] | ||
| = Null Payloads and Log Compaction of 'Tombstone' Records | ||
|
|
||
| When using log compaction, you can send and receive messages with `null` payloads to identify the deletion of a key. | ||
| You can also receive `null` values for other reasons, such as a deserializer that might return `null` when it cannot deserialize a value. | ||
|
|
||
| [[tombstones.produce]] | ||
| == Producing Null Payloads | ||
| To send a `null` payload by using the `PulsarTemplate`, you can use the fluent API and pass null into the value argument of the `newMessage()` method, for example: | ||
| [source, java] | ||
| ---- | ||
| pulsarTemplate | ||
| .newMessage(null) | ||
| .withTopic("my-topic") | ||
| .withSchema(Schema.STRING) | ||
| .withMessageCustomizer((mb) -> mb.key("key:1234")) | ||
| .send(); | ||
| ---- | ||
| NOTE: When sending null values you must specify the schema type as the system can not determine the type of the message from a `null` payload. | ||
|
|
||
| [[tombstones.consume]] | ||
| == Consuming Null Payloads | ||
| For `@PulsarListener` and `@PulsarReader`, the `null` payload is passed into the listener method based on the type of its message parameter as follows: | ||
| |=== | ||
| | Parameter type | Passed-in value | ||
|
|
||
| | primitive | ||
| | `null` | ||
|
|
||
| | user-defined | ||
| | `null` | ||
|
|
||
| | `org.apache.pulsar.client.api.Message<T>` | ||
| | non-null Pulsar message whose `getValue()` returns `null` | ||
|
|
||
| | `org.springframework.messaging.Message<T>` | ||
| | non-null Spring message whose `getPayload()` returns `PulsarNull` | ||
|
|
||
| | `List<X>` | ||
| | non-null list whose entries (`X`) are one of the above types and act accordingly (ie. primitive entries are `null` etc..) | ||
|
|
||
| | `org.apache.pulsar.client.api.Messages<T>` | ||
| | non-null container of non-null Pulsar messages whose `getValue()` returns `null` | ||
|
|
||
| |=== | ||
|
|
||
| IMPORTANT: When the passed-in value is `null` (ie. single record listeners with primitive or user-defined types) you must use the `@Payload` parameter annotation with `required = false`. | ||
|
|
||
| IMPORTANT: When using the Spring `org.springframework.messaging.Message` for your listener payload type, its generic type information must be wide enough to accept `Message<PulsarNull>` (eg. `Message`, `Message<?>`, or `Message<Object>`). | ||
| This is due to the fact that the Spring Message does not allow null values for its payload and instead uses the `PulsarNull` placeholder. | ||
|
|
||
| If it is a tombstone message for a compacted log, you usually also need the key so that your application can determine which key was +++"+++`deleted`+++"+++. | ||
| The following example shows such a configuration: | ||
|
|
||
| [source, java] | ||
| ---- | ||
| @PulsarListener( | ||
| topics = "my-topic", | ||
| subscriptionName = "my-topic-sub", | ||
| schemaType = SchemaType.STRING) | ||
| void myListener( | ||
| @Payload(required = false) String msg, | ||
| @Header(PulsarHeaders.KEY) String key) { | ||
| ... | ||
| } | ||
| ---- | ||
|
|
||
| NOTE: The `@PulsarReader` does not yet support `@Header` arguments, so it is less useful in the log compaction scenario. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.