You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gw...@apache.org on 2016/11/02 17:27:22 UTC
kafka git commit: MINOR: Fix NPE when Connect offset contains
non-primitive type
Repository: kafka
Updated Branches:
refs/heads/trunk 2959bc2ad -> 508711d09
MINOR: Fix NPE when Connect offset contains non-primitive type
When storing a non-primitive type in a Connect offset, the following NullPointerException will occur:
```
07:18:23.702 [pool-3-thread-1] ERROR o.a.k.c.storage.OffsetStorageWriter - CRITICAL: Failed to serialize offset data, making it impossible to commit offsets under namespace tenant-db-bootstrap-source. This likely won't recover unless the unserializable partition or offset information is overwritten.
07:18:23.702 [pool-3-thread-1] ERROR o.a.k.c.storage.OffsetStorageWriter - Cause of serialization failure:
java.lang.NullPointerException: null
at org.apache.kafka.connect.storage.OffsetUtils.validateFormat(OffsetUtils.java:51)
at org.apache.kafka.connect.storage.OffsetStorageWriter.doFlush(OffsetStorageWriter.java:143)
at org.apache.kafka.connect.runtime.WorkerSourceTask.commitOffsets(WorkerSourceTask.java:319)
... snip ...
```
The attached patch fixes the specific case where OffsetUtils.validateFormat is attempting to provide a useful error message, but fails to because the schemaType method could return null.
This contribution is my original work and I license the work to the project under the project's open source license.
Author: Mathieu Fenniak <ma...@replicon.com>
Reviewers: Gwen Shapira
Closes #2087 from mfenniak/fix-npr-with-clearer-error-message
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/508711d0
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/508711d0
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/508711d0
Branch: refs/heads/trunk
Commit: 508711d09e7c6070868aadfff1b2627e77f7f3d9
Parents: 2959bc2
Author: Mathieu Fenniak <ma...@replicon.com>
Authored: Wed Nov 2 10:27:19 2016 -0700
Committer: Gwen Shapira <cs...@gmail.com>
Committed: Wed Nov 2 10:27:19 2016 -0700
----------------------------------------------------------------------
.../main/java/org/apache/kafka/connect/storage/OffsetUtils.java | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/508711d0/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetUtils.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetUtils.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetUtils.java
index b457b12..9fdcfc3 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetUtils.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetUtils.java
@@ -48,6 +48,8 @@ public class OffsetUtils {
if (value == null)
continue;
Schema.Type schemaType = ConnectSchema.schemaType(value.getClass());
+ if (schemaType == null)
+ throw new DataException("Offsets may only contain primitive types as values, but field " + entry.getKey() + " contains " + value.getClass());
if (!schemaType.isPrimitive())
throw new DataException("Offsets may only contain primitive types as values, but field " + entry.getKey() + " contains " + schemaType);
}