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 2015/10/24 02:01:44 UTC
kafka git commit: KAFKA-2626: Handle null keys and value validation
properly in OffsetStorageWriter.
Repository: kafka
Updated Branches:
refs/heads/trunk 2e6177359 -> 6f2f1f984
KAFKA-2626: Handle null keys and value validation properly in OffsetStorageWriter.
Author: Ewen Cheslack-Postava <me...@ewencp.org>
Reviewers: Gwen Shapira
Closes #345 from ewencp/kafka-2626-offset-storage-writer-null-values
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/6f2f1f98
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/6f2f1f98
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/6f2f1f98
Branch: refs/heads/trunk
Commit: 6f2f1f9843f537b9bda3aa3951a867fdee661761
Parents: 2e61773
Author: Ewen Cheslack-Postava <me...@ewencp.org>
Authored: Fri Oct 23 17:01:33 2015 -0700
Committer: Gwen Shapira <cs...@gmail.com>
Committed: Fri Oct 23 17:01:33 2015 -0700
----------------------------------------------------------------------
.../kafka/copycat/storage/OffsetUtils.java | 5 ++
.../storage/OffsetStorageWriterTest.java | 71 +++++++++++++++-----
2 files changed, 59 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/6f2f1f98/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetUtils.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetUtils.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetUtils.java
index 8d78a57..9ba7662 100644
--- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetUtils.java
+++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetUtils.java
@@ -34,6 +34,11 @@ public class OffsetUtils {
}
public static <K, V> void validateFormat(Map<K, V> offsetData) {
+ // Both keys and values for offsets may be null. For values, this is a useful way to delete offsets or indicate
+ // that there's not usable concept of offsets in your source system.
+ if (offsetData == null)
+ return;
+
for (Map.Entry<K, V> entry : offsetData.entrySet()) {
if (!(entry.getKey() instanceof String))
throw new DataException("Offsets may only use String keys");
http://git-wip-us.apache.org/repos/asf/kafka/blob/6f2f1f98/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/OffsetStorageWriterTest.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/OffsetStorageWriterTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/OffsetStorageWriterTest.java
index e33ecd0..3dd0b52 100644
--- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/OffsetStorageWriterTest.java
+++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/OffsetStorageWriterTest.java
@@ -45,15 +45,11 @@ public class OffsetStorageWriterTest {
private static final String NAMESPACE = "namespace";
// Copycat format - any types should be accepted here
private static final Map<String, String> OFFSET_KEY = Collections.singletonMap("key", "key");
- private static final List<Object> OFFSET_KEY_WRAPPED = Arrays.asList(NAMESPACE, OFFSET_KEY);
private static final Map<String, Integer> OFFSET_VALUE = Collections.singletonMap("key", 12);
// Serialized
private static final byte[] OFFSET_KEY_SERIALIZED = "key-serialized".getBytes();
private static final byte[] OFFSET_VALUE_SERIALIZED = "value-serialized".getBytes();
- private static final Map<ByteBuffer, ByteBuffer> OFFSETS_SERIALIZED
- = Collections.singletonMap(ByteBuffer.wrap(OFFSET_KEY_SERIALIZED),
- ByteBuffer.wrap(OFFSET_VALUE_SERIALIZED));
@Mock private OffsetBackingStore store;
@Mock private Converter keyConverter;
@@ -79,7 +75,7 @@ public class OffsetStorageWriterTest {
public void testWriteFlush() throws Exception {
@SuppressWarnings("unchecked")
Callback<Void> callback = PowerMock.createMock(Callback.class);
- expectStore(callback, false);
+ expectStore(OFFSET_KEY, OFFSET_KEY_SERIALIZED, OFFSET_VALUE, OFFSET_VALUE_SERIALIZED, callback, false, null);
PowerMock.replayAll();
@@ -91,6 +87,41 @@ public class OffsetStorageWriterTest {
PowerMock.verifyAll();
}
+ // It should be possible to set offset values to null
+ @Test
+ public void testWriteNullValueFlush() throws Exception {
+ @SuppressWarnings("unchecked")
+ Callback<Void> callback = PowerMock.createMock(Callback.class);
+ expectStore(OFFSET_KEY, OFFSET_KEY_SERIALIZED, null, null, callback, false, null);
+
+ PowerMock.replayAll();
+
+ writer.offset(OFFSET_KEY, null);
+
+ assertTrue(writer.beginFlush());
+ writer.doFlush(callback).get(1000, TimeUnit.MILLISECONDS);
+
+ PowerMock.verifyAll();
+ }
+
+ // It should be possible to use null keys. These aren't actually stored as null since the key is wrapped to include
+ // info about the namespace (connector)
+ @Test
+ public void testWriteNullKeyFlush() throws Exception {
+ @SuppressWarnings("unchecked")
+ Callback<Void> callback = PowerMock.createMock(Callback.class);
+ expectStore(null, null, OFFSET_VALUE, OFFSET_VALUE_SERIALIZED, callback, false, null);
+
+ PowerMock.replayAll();
+
+ writer.offset(null, OFFSET_VALUE);
+
+ assertTrue(writer.beginFlush());
+ writer.doFlush(callback).get(1000, TimeUnit.MILLISECONDS);
+
+ PowerMock.verifyAll();
+ }
+
@Test
public void testNoOffsetsToFlush() {
// If no offsets are flushed, we should finish immediately and not have made any calls to the
@@ -112,9 +143,9 @@ public class OffsetStorageWriterTest {
@SuppressWarnings("unchecked")
final Callback<Void> callback = PowerMock.createMock(Callback.class);
// First time the write fails
- expectStore(callback, true);
+ expectStore(OFFSET_KEY, OFFSET_KEY_SERIALIZED, OFFSET_VALUE, OFFSET_VALUE_SERIALIZED, callback, true, null);
// Second time it succeeds
- expectStore(callback, false);
+ expectStore(OFFSET_KEY, OFFSET_KEY_SERIALIZED, OFFSET_VALUE, OFFSET_VALUE_SERIALIZED, callback, false, null);
// Third time it has no data to flush so we won't get past beginFlush()
PowerMock.replayAll();
@@ -135,7 +166,7 @@ public class OffsetStorageWriterTest {
final Callback<Void> callback = PowerMock.createMock(Callback.class);
// Trigger the send, but don't invoke the callback so we'll still be mid-flush
CountDownLatch allowStoreCompleteCountdown = new CountDownLatch(1);
- expectStore(null, false, allowStoreCompleteCountdown);
+ expectStore(OFFSET_KEY, OFFSET_KEY_SERIALIZED, OFFSET_VALUE, OFFSET_VALUE_SERIALIZED, null, false, allowStoreCompleteCountdown);
PowerMock.replayAll();
@@ -165,7 +196,7 @@ public class OffsetStorageWriterTest {
CountDownLatch allowStoreCompleteCountdown = new CountDownLatch(1);
// In this test, the write should be cancelled so the callback will not be invoked and is not
// passed to the expectStore call
- expectStore(null, false, allowStoreCompleteCountdown);
+ expectStore(OFFSET_KEY, OFFSET_KEY_SERIALIZED, OFFSET_VALUE, OFFSET_VALUE_SERIALIZED, null, false, allowStoreCompleteCountdown);
PowerMock.replayAll();
@@ -180,13 +211,13 @@ public class OffsetStorageWriterTest {
PowerMock.verifyAll();
}
- private void expectStore(final Callback<Void> callback, final boolean fail) {
- expectStore(callback, fail, null);
- }
-
/**
* Expect a request to store data to the underlying OffsetBackingStore.
*
+ * @param key the key for the offset
+ * @param keySerialized serialized version of the key
+ * @param value the value for the offset
+ * @param valueSerialized serialized version of the value
* @param callback the callback to invoke when completed, or null if the callback isn't
* expected to be invoked
* @param fail if true, treat
@@ -195,14 +226,20 @@ public class OffsetStorageWriterTest {
* ensure tests complete.
* @return the captured set of ByteBuffer key-value pairs passed to the storage layer
*/
- private void expectStore(final Callback<Void> callback,
+ private void expectStore(Map<String, String> key, byte[] keySerialized,
+ Map<String, Integer> value, byte[] valueSerialized,
+ final Callback<Void> callback,
final boolean fail,
final CountDownLatch waitForCompletion) {
- EasyMock.expect(keyConverter.fromCopycatData(NAMESPACE, null, OFFSET_KEY_WRAPPED)).andReturn(OFFSET_KEY_SERIALIZED);
- EasyMock.expect(valueConverter.fromCopycatData(NAMESPACE, null, OFFSET_VALUE)).andReturn(OFFSET_VALUE_SERIALIZED);
+ List<Object> keyWrapped = Arrays.asList(NAMESPACE, key);
+ EasyMock.expect(keyConverter.fromCopycatData(NAMESPACE, null, keyWrapped)).andReturn(keySerialized);
+ EasyMock.expect(valueConverter.fromCopycatData(NAMESPACE, null, value)).andReturn(valueSerialized);
final Capture<Callback<Void>> storeCallback = Capture.newInstance();
- EasyMock.expect(store.set(EasyMock.eq(OFFSETS_SERIALIZED), EasyMock.capture(storeCallback)))
+ final Map<ByteBuffer, ByteBuffer> offsetsSerialized = Collections.singletonMap(
+ keySerialized == null ? null : ByteBuffer.wrap(keySerialized),
+ valueSerialized == null ? null : ByteBuffer.wrap(valueSerialized));
+ EasyMock.expect(store.set(EasyMock.eq(offsetsSerialized), EasyMock.capture(storeCallback)))
.andAnswer(new IAnswer<Future<Void>>() {
@Override
public Future<Void> answer() throws Throwable {