[
https://issues.apache.org/jira/browse/KAFKA-6378?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16484502#comment-16484502
]
ASF GitHub Bot commented on KAFKA-6378:
---------------------------------------
wicknicks opened a new pull request #5065: KAFKA-6378: Implement error handling
for source and sink tasks
URL: https://github.com/apache/kafka/pull/5065
This PR implements the features described in this KIP:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-298%3A+Error+Handling+in+Connect
This PR changes the Connect framework to allow it to automatically deal with
errors encountered while processing records in a Connector. The following
behavior changes are introduced here:
**Retry on Failure**: Retry the failed operation a configurable number of
times, with backoff between each retry.
**Task Tolerance Limits**: Tolerate a configurable number of failures in a
task.
We also add the following ways to report errors, along with sufficient
context to simplify the debugging process:
**Log Error Context**: The error information along with processing context
is logged along with standard application logs.
**Dead Letter Queue**: Produce the original message into a Kafka topic
(applicable only to sink connectors).
The logged information consists of the following bits:
New **metrics** which will monitor the number of failures, and the behavior
of the response handler are added.
The changes proposed here **are backward compatible**. The current behavior
in Connect is to kill the task on the first error in any stage. This will
remain the default behavior if the connector does not override any of the new
configurations which are provided as part of this feature.
Testing: added multiple unit tests to test the retry and tolerance logic.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> NullPointerException on KStream-GlobalKTable leftJoin when KeyValueMapper
> returns null
> --------------------------------------------------------------------------------------
>
> Key: KAFKA-6378
> URL: https://issues.apache.org/jira/browse/KAFKA-6378
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 1.0.0
> Reporter: Andy Bryant
> Assignee: Andy Bryant
> Priority: Major
> Fix For: 1.1.0, 2.0.0, 1.0.1
>
>
> On a Stream->GlobalKTable leftJoin if the KeyValueMapper returns null, the
> stream fails with a NullPointerException (see stacktrace below). On Kafka
> 0.11.0.0 the stream processes this successfully, calling the ValueJoiner with
> the table value set to null.
> The use-case for this is joining a stream to a table containing reference
> data where the stream foreign key may be null. There is no straight-forward
> workaround in this case with Kafka 1.0.0 without having to resort to either
> generating a key that will never match or branching the stream for records
> that don't have the foreign key.
> Exception in thread "workshop-simple-example-client-StreamThread-1"
> java.lang.NullPointerException
> at java.base/java.util.Objects.requireNonNull(Objects.java:221)
> at
> org.apache.kafka.streams.state.internals.CachingKeyValueStore.get(CachingKeyValueStore.java:136)
> at
> org.apache.kafka.streams.state.internals.CachingKeyValueStore.get(CachingKeyValueStore.java:35)
> at
> org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore.get(InnerMeteredKeyValueStore.java:184)
> at
> org.apache.kafka.streams.state.internals.MeteredKeyValueBytesStore.get(MeteredKeyValueBytesStore.java:116)
> at
> org.apache.kafka.streams.kstream.internals.KTableSourceValueGetterSupplier$KTableSourceValueGetter.get(KTableSourceValueGetterSupplier.java:49)
> at
> org.apache.kafka.streams.kstream.internals.KStreamKTableJoinProcessor.process(KStreamKTableJoinProcessor.java:56)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:46)
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:124)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:85)
> at
> org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:80)
> at
> org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:216)
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.process(AssignedTasks.java:403)
> at
> org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:317)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.processAndMaybeCommit(StreamThread.java:942)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:822)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
