DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: here [Change the link from KAFKA-1 to your own ticket]
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
- How it works
- Upload the payload to external storage (e.g., S3, HDFS, or a database).
- Send a reference (e.g.,
s3://bucket/key) via Kafka. - Consumers fetch the payload using the reference
- Considerations
- Reduces Kafka network/storage load.
- Introduces dependency on external systems.
- Available open-source implementation
- There isn't a specific open-source implementation for Reference-Based Messaging. Mostly everyone adopting this has their custom solution
| Pattern | Pros | Cons |
|---|---|---|
| Chunking | No external storage required | Complex client logic to split and reassemble messages |
| Reference-Based | Minimizes Kafka load | External system dependency |
This KIP is proposing
- A serializer in Apache Kafka that implements Reference-Based Messaging as this is the simplest one.
- A notion of a
Composableserializer where the client can apply a list of serializers before applying a large message serializer
...
1. Composable Serializer/Deserializer
Configuration
| Key | Description | Default | Required |
| value.serializers | List of serializers class that implements the `org.apache.kafka.common.serialization.Serializer` interface in order. The first serializer is `Serializer<T>` while the remaining serializers must be `Serializer<byte[]>`. | None | Yes |
| value.deserializers | List of deserializes class that implements the `org.apache.kafka.common.serialization.Deserializer` interface in order. The last deserializer must be `Deserializer<T>` while the rest must be `Deserializer<byte[]>`. | None | Yes |
2. LargeMessageSerializer
Configuration
| Key | Description | Default | Required |
|---|---|---|---|
large.message.blob.store.class | Implementation of org.apache.kafka.common.serialization.largemessage.blobstore.BlobStore . | None | Yes |
large.message.blob.store.timeout.ms | Timeout for blob store operations. This can't exceedmax.block.msfor Kafka producers ormax.poll.interval.msin Kafka consumer. | 10000 | No |
large.message.blob.store.retry.count | Number of retries for blob store operations. | 5 | No |
large.message.blob.store.retry.max.backoff.ms | Backoff time between retries for blob store operations. | 10000 | No |
large.message.blob.store.retry.delay.backoff.ms | Delay time between retries back off for blob store operations. | 100 | No |
large.message.blob.store.blob.id.generator | Implementation of org.apache.kafka.common.serialziation.largemessage.blobstore.BlobIdGenerator | org.apache.kafka.common.serialization.largemessage.blobstore.BlobIdGenerator.DefaultBlobIDGenerator | No |
large.message.threshold.bytes | The maximum size of the message is considered large. | 1MB (Default of max.message.bytes) | No |
3. LargeMessageDeserializer
Configuration
| Key | Description | Default | Required |
|---|---|---|---|
large.message.blob.store.class | Implementation of org.apache.kafka.common.serialization.largemessage.blobstore.BlobStore . | None | Yes |
large.message.blob.store.timeout.ms | Timeout for blob store operations. This can't exceedmax.block.msfor Kafka producers ormax.poll.interval.msin Kafka consumer. | 10000 | No |
large.message.blob.store.retry.count | Number of retries for blob store operations. | 5 | No |
large.message.blob.store.retry.max.backoff.ms | Backoff time between retries for blob store operations. | 10000 | No |
large.message.blob.store.retry.delay.backoff.ms | Delay time between retries back off for blob store operations. | 100 | No |
| Implementation of org.apache.kafka.common.serialziation.largemessage.blobstore.BlobIdGenerator iinterface. This provides a way for the serializer to generate an identifier for the message in the store. | org.apache.kafka.common.serialization.largemessage.blobstore.BlobIdGenerator.DefaultBlobIDGenerator | No |
| `large.message.skip.not.found.error` | Skip not found error when the blob is not found in the store. This allows the deserializer to skip not-found messages and return empty bytes instead. This is important when external store ttl is smaller than kafka retention. | FALSE | No |
4. BlobStore
| Code Block | ||
|---|---|---|
| ||
public abstract class BlobStore implements Configurable, Closeable {
@Override
public void configure(Map<String, ?> configs) {
// configure
}
/**
* Upload data into object store.
*
* @param data data that will be uploaded into the object.
* @return {@link BlobResponse}.
*/
public abstract BlobResponse putObject(String topic, byte[] data);
/**
* Download object from object store.
*
* @param blobPath id of the object in blobstore
* @return content of the object as bytes.
*/
public abstract byte[] getObject(String blobPath);
} |
...
6. LargeBlobMessage
| Code Block | ||
|---|---|---|
| ||
public interface LargeMessageFormatter<T> extends Configurable {
LargeMessageFormatter JSON = new LargeMessageFormatter() {
// implement a default json one
// format is {
// "full-blob-path": "<full-path-to-access-blob-as-string>",
// "blob-store-class": "<used-blob-store-class-path-to-upload-blob>",
// "blob-id-generator-class": "<class-used-to-generate-blob-id>",
// "large-message-formatter-class": "<formatter-class-path>"
// }
};
/**
* build message bytes from blob store response and data.
*/
byte[] messageBytes(byte[] data, BlobResponse response);
/**
* parse data into LargeBlobMessage.
*/
T largeBlobMessage(byte[] data);
} |
...