DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
| Code Block | ||
|---|---|---|
| ||
public interface ProductionExceptionHandler extends Configurable {
... /**
@Deprecated
default ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,
final ProducerRecord<byte[], byte[]> record,
final Exception exception) {
throw new UnsupportedOperationException();
}
/**
* Inspect a record that we attempted to produce, and the exception that resulted
* from attempting to produce it and determine to continue or stop processing.
*
* @param context
* The error handler context metadata.
* @param record
* The record that failed to produce.
* @param exception
* The exception that occurred during production.
*
* @return a {@link ProductionExceptionResponse} object
*/
default ProductionExceptionResponse handleError(final ErrorHandlerContext context,
final ProducerRecord<byte[], byte[]> record,
final Exception exception) {
final ProductionExceptionHandlerResponse response = handle(context, record, exception);
if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL == response) {
return ProductionExceptionResponse.failProcessing();
} else if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.RETRY == response) {
return ProductionExceptionResponse.retryProcessing();
}
return ProductionExceptionResponse.continueProcessing();
return new ProductionExceptionResponse(handle(context, record, exception), Collections.emptyList());
}
@Deprecated
default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
final Exception exception) {
return ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL;
}
/**
* Handles serialization exception and determine if the process should continue. The default implementation is to
* fail the process.
*
* @param context
* The error handler context metadata.
* @param record
* The record that failed to serialize.
* @param exception
* The exception that occurred during serialization.
* @param origin
* The origin of the serialization exception.
*
* @return a {@link ProductionExceptionResponse} object
*/
default ProductionExceptionResponse handleSerializationError(final ErrorHandlerContext context,
* The origin of the serialization exception.
*
* @return a {@link ProductionExceptionResponse} object
*/
default ProductionExceptionResponse handleSerializationError(final ProducerRecordErrorHandlerContext recordcontext,
final ExceptionProducerRecord exceptionrecord,
final SerializationExceptionOrigin origin) {Exception exception,
final ProductionExceptionHandlerResponse response = handleSerializationException(context, record, exception, origin);
if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL == response) {
return ProductionExceptionResponse.failProcessing();
} else if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.RETRY == response) {
final return ProductionExceptionResponse.retryProcessing();
SerializationExceptionOrigin origin) {
}
return new ProductionExceptionResponse(handleSerializationException(context, record, exception, returnorigin), ProductionExceptionResponseCollections.continueProcessingemptyList());
}
...
/**
* Represents the result of handling a production exception.
* <p>
* The {@code Response} class encapsulates a {@link ProductionExceptionHandlerResponse},
* indicating whether processing should continue or fail, along with an optional list of
* {@link ProducerRecord} instances to be sent to a dead letter queue.
* </p>
*/
class ProductionExceptionResponse {
private ProductionExceptionHandlerResponse productionExceptionHandlerResponse;
private List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords;
/**
* Constructs a new {@code ProductionExceptionResponse} object.
*
* @param productionExceptionHandlerResponse the response indicating whether processing should continue or fail;
* must not be {@code null}.
* @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
*/
private ProductionExceptionResponse(final ProductionExceptionHandlerResponse productionExceptionHandlerResponse,
final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
this.productionExceptionHandlerResponse = productionExceptionHandlerResponse;
this.deadLetterQueueRecords = deadLetterQueueRecords;
}
/**
* Creates a {@code ProductionExceptionResponse} indicating that processing should fail.
*
* @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
* @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#FAIL} status.
*/
public static ProductionExceptionResponse failProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
return new ProductionExceptionResponse(ProductionExceptionHandlerResponse.FAIL, deadLetterQueueRecords);
}
/**
* Creates a {@code ProductionExceptionResponse} indicating that processing should fail.
*
* @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#FAIL} status.
*/
public static ProductionExceptionResponse failProcessing() {
return failProcessing(Collections.emptyList());
}
/**
* Creates a {@code ProductionExceptionResponse} indicating that processing should continue.
*
* @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
* @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.
*/
public static ProductionExceptionResponse continueProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
return new ProductionExceptionResponse(ProductionExceptionHandlerResponse.CONTINUE, deadLetterQueueRecords);
}
/**
* Creates a {@code ProductionExceptionResponse} indicating that processing should continue.
*
* @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.
*/
public static ProductionExceptionResponse continueProcessing() {
return continueProcessing(Collections.emptyList());
}
/**
* Creates a {@code ProductionExceptionResponse} indicating that processing should retry.
*
* @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.
*/
public static ProductionExceptionResponse retryProcessing() {
return new ProductionExceptionResponse(ProductionExceptionHandlerResponse.RETRY, Collections.emptyList());
}
/**
* Retrieves the production exception handler response.
*
* @return the {@link ProductionExceptionHandlerResponse} indicating whether processing should continue or fail.
*/
public ProductionExceptionHandlerResponse response() {
return productionExceptionHandlerResponse;
}
/**
* Retrieves an unmodifiable list of records to be sent to the dead letter queue.
* <p>
* If the list is {@code null}, an empty list is returned.
* </p>
*
* @return an unmodifiable list of {@link ProducerRecord} instances
* for the dead letter queue, or an empty list if no records are available.
*/
public List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
if (deadLetterQueueRecords == null) {
return Collections.emptyList();
}
return Collections.unmodifiableList(deadLetterQueueRecords);
}
}
...
}
|
...
| Code Block | ||
|---|---|---|
| ||
public interface DeserializationExceptionHandler extends Configurable {
...
@Deprecated
default DeserializationHandlerResponse handle(final ErrorHandlerContext context,
final ConsumerRecord<byte[], byte[]> record,
final Exception exception) {
throw new UnsupportedOperationException();
}
/**
* Inspects a record and the exception received during deserialization.
*
* @param context
* Error handler context.
* @param record
* Record that failed deserialization.
* @param exception
* The actual exception.
*
* @return a {@link DeserializationExceptionResponse} object
*/
default DeserializationExceptionResponse handleError(final ErrorHandlerContext context, final ConsumerRecord<byte[], byte[]> record, final Exception exception) {
if (DeserializationHandlerResponse.FAIL == handle(context, record, exception)) {
return DeserializationExceptionResponse.failProcessing(Collections.emptyList());
}
return DeserializationExceptionResponse.continueProcessing( final Exception exception) { return new DeserializationExceptionResponse(handle(context, record, exception), Collections.emptyList());
}
...
/**
* Represents the result of handling a deserialization exception.
* <p>
* The {@code Response} class encapsulates a {@link ProcessingExceptionHandler.ProcessingHandlerResponse},
* indicating whether processing should continue or fail, along with an optional list of
* {@link ProducerRecord} instances to be sent to a dead letter queue.
* </p>
*/
class DeserializationExceptionResponse {
private DeserializationHandlerResponse deserializationHandlerResponse;
private List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords;
/**
* Constructs a new {@code DeserializationExceptionResponse} object.
*
* @param deserializationHandlerResponse the response indicating whether processing should continue or fail;
* must not be {@code null}.
* @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
*/
private DeserializationExceptionResponse(final DeserializationHandlerResponse deserializationHandlerResponse,
final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
this.deserializationHandlerResponse = deserializationHandlerResponse;
this.deadLetterQueueRecords = deadLetterQueueRecords;
}
/**
* Creates a {@code DeserializationExceptionResponse} indicating that processing should fail.
*
* @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
* @return a {@code DeserializationExceptionResponse} with a {@link DeserializationHandlerResponse#FAIL} status.
*/
public static DeserializationExceptionResponse failProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
return new DeserializationExceptionResponse(DeserializationHandlerResponse.FAIL, deadLetterQueueRecords);
}
/**
* Creates a {@code DeserializationExceptionResponse} indicating that processing should fail.
*
* @return a {@code DeserializationExceptionResponse} with a {@link DeserializationHandlerResponse#FAIL} status.
*/
public static DeserializationExceptionResponse failProcessing() {
return failProcessing(Collections.emptyList());
}
/**
* Creates a {@code DeserializationExceptionResponse} indicating that processing should continue.
*
* @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
* @return a {@code DeserializationExceptionResponse} with a {@link DeserializationHandlerResponse#CONTINUE} status.
*/
public static DeserializationExceptionResponse continueProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
return new DeserializationExceptionResponse(DeserializationHandlerResponse.CONTINUE, deadLetterQueueRecords);
}
/**
* Creates a {@code DeserializationExceptionResponse} indicating that processing should continue.
*
* @return a {@code DeserializationExceptionResponse} with a {@link DeserializationHandlerResponse#CONTINUE} status.
*/
public static DeserializationExceptionResponse continueProcessing() {
return continueProcessing(Collections.emptyList());
}
/**
* Retrieves the deserialization handler response.
*
* @return the {@link DeserializationHandlerResponse} indicating whether processing should continue or fail.
*/
public DeserializationHandlerResponse response() {
return deserializationHandlerResponse;
}
/**
* Retrieves an unmodifiable list of records to be sent to the dead letter queue.
* <p>
* If the list is {@code null}, an empty list is returned.
* </p>
*
* @return an unmodifiable list of {@link ProducerRecord} instances
* for the dead letter queue, or an empty list if no records are available.
*/
public List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
if (deadLetterQueueRecords == null) {
return Collections.emptyList();
}
return Collections.unmodifiableList(deadLetterQueueRecords);
}
}
}
|
...
| Code Block | ||
|---|---|---|
| ||
public interface ProcessingExceptionHandler extends Configurable {
...
@Deprecated
default ProcessingHandlerResponse handle(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception){
throw new UnsupportedOperationException();
};
/**
* Inspects a record and the exception received during processing.
*
* @param context
* Processing context metadata.
* @param record
* Record where the exception occurred.
* @param exception
* The actual exception.
*
* @return a {@link ProcessingExceptionResponse} object
*/
default ProcessingExceptionResponse handleError(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {
if (ProcessingHandlerResponse.FAIL == handle(context, record, exception)) {
return new return ProcessingExceptionResponse.failProcessing();
}
return ProcessingExceptionResponse.continueProcessing();
ProcessingExceptionResponse(handle(context, record, exception), Collections.emptyList());
}
...
/**
* Represents the result of handling a processing exception.
* <p>
* The {@code Response} class encapsulates a {@link ProcessingHandlerResponse},
* indicating whether processing should continue or fail, along with an optional list of
* {@link org.apache.kafka.clients.producer.ProducerRecord} instances to be sent to a dead letter queue.
* </p>
*/
class ProcessingExceptionResponse {
private ProcessingHandlerResponse processingHandlerResponse;
private List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords;
/**
* Constructs a new {@code ProcessingExceptionResponse} object.
*
* @param processingHandlerResponse the response indicating whether processing should continue or fail;
* must not be {@code null}.
* @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
*/
private ProcessingExceptionResponse(final ProcessingHandlerResponse processingHandlerResponse,
final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
this.processingHandlerResponse = processingHandlerResponse;
this.deadLetterQueueRecords = deadLetterQueueRecords;
}
/**
* Creates a {@code ProcessingExceptionResponse} indicating that processing should fail.
*
* @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
* @return a {@code ProcessingExceptionResponse} with a {@link ProcessingHandlerResponse#FAIL} status.
*/
public static ProcessingExceptionResponse failProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
return new ProcessingExceptionResponse(ProcessingHandlerResponse.FAIL, deadLetterQueueRecords);
}
/**
* Creates a {@code ProcessingExceptionResponse} indicating that processing should fail.
*
* @return a {@code ProcessingExceptionResponse} with a {@link ProcessingHandlerResponse#FAIL} status.
*/
public static ProcessingExceptionResponse failProcessing() {
return failProcessing(Collections.emptyList());
}
/**
* Creates a {@code ProcessingExceptionResponse} indicating that processing should continue.
*
* @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
* @return a {@code Response} with a {@link ProcessingHandlerResponse#CONTINUE} status.
*/
public static ProcessingExceptionResponse continueProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
return new ProcessingExceptionResponse(ProcessingHandlerResponse.CONTINUE, deadLetterQueueRecords);
}
/**
* Creates a {@code ProcessingExceptionResponse} indicating that processing should continue.
*
* @return a {@code ProcessingExceptionResponse} with a {@link ProcessingHandlerResponse#CONTINUE} status.
*/
public static ProcessingExceptionResponse continueProcessing() {
return continueProcessing(Collections.emptyList());
}
/**
* Retrieves the processing handler response.
*
* @return the {@link ProcessingHandlerResponse} indicating whether processing should continue or fail.
*/
public ProcessingHandlerResponse response() {
return processingHandlerResponse;
}
/**
* Retrieves an unmodifiable list of records to be sent to the dead letter queue.
* <p>
* If the list is {@code null}, an empty list is returned.
* </p>
*
* @return an unmodifiable list of {@link ProducerRecord} instances
* for the dead letter queue, or an empty list if no records are available.
*/
public List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
if (deadLetterQueueRecords == null) {
return Collections.emptyList();
}
return Collections.unmodifiableList(deadLetterQueueRecords);
}
}
}
|
...