DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
| Code Block | ||
|---|---|---|
| ||
/**
* Response from publish / download from PayloadStore back to the serialization layer
* It contains the final path, response code and the encountered exception if there was any.
*/
public class PayloadResponse {
public final int responseCode;
public final String path;
public final PayloadStoreException payloadStoreException;
public final Boolean retriable;
/**
* Construct payload response with response code and payload id.
*/
public PayloadResponse(int responseCode, String path) {
this(responseCode, path, null, false);
}
/**
* Construct payload response with response code, payload id and exception.
*/
public PayloadResponse(int responseCode, String path, PayloadStoreExceptionPayloadSto
reException payloadStoreException, Boolean retriable) {
this.responseCode = responseCode;
this.fullPayloadPath = path;
this.payloadStoreException = payloadStoreException;
this.retriable = retriable;
}
} |
8. PayloadStoreException
| Code Block |
|---|
/**
* Exception class that can either be reliable or not
* this helps the serializer/desrializer to decided either to retry or to crash.
* One subclass will be added is PayloadNotFoundException which is used to indicated if the payload not found
* This is used by deserializer to skip or not.
**/
public class PayloadStoreException extends RuntimeException {
protected boolean isRetryable = true;
/**
* Constructor PayloadStoreException with message and throwable.
*/
public PayloadStoreException(String message, Throwable t) {
super(message, t);
}
/**
* Constructor PayloadStoreException with message.
*/
public PayloadStoreException(String message) {
super(message);
}
/**
* Constructor PayloadStoreException with throwable.
*/
public PayloadStoreException(Throwable t) {
super(t);
}
/**
* Constructor PayloadStoreException with message, throwable and if it is retryable or not.
*/
public PayloadStoreException(String message, Throwable t, boolean retryable) {
this(message, t);
isRetryable = retryable;
}
/**
* return whether the exception is retryable or not.
*/
public boolean isRetryable() {
return isRetryable;
}
}
|
Example
| Code Block | ||
|---|---|---|
| ||
Map<String, Object> producerConfig = new HashMap<>();
producerConfig.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerConfig.put("value.serializers",
"org.apache.kafka.common.serialization.DoubleSerializer,org.apache.kafka.common.serialization.LargeMessageSerializer"); producerConfig.put("large.message.payload.store.class", "CustomS3Store")
producerConfig.put("s3.bucket", "my-bucket")
producerConfig.put("bootstrap.servers", "localhost:9092");
KafkaProducer<String, Double> producer = new KafkaProducer<>(producerConfig); |
...