You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2017/06/22 00:16:51 UTC
kafka git commit: KAFKA-5474: Streams StandbyTask should no
checkpoint on commit if EOS is enabled
Repository: kafka
Updated Branches:
refs/heads/trunk bc47e9d6c -> 914e42a28
KAFKA-5474: Streams StandbyTask should no checkpoint on commit if EOS is enabled
<strike> - actual fix for `StandbyTask#commit()` </strike>
Additionally (for debugging):
- EOS test, does not report "expected" value correctly
- add `IntegerDecoder` (to be use with `kafka.tools.DumpLogSegments`)
- add test for `StreamTask` to not checkpoint on commit if EOS enabled
Author: Matthias J. Sax <ma...@confluent.io>
Reviewers: Bill Bejeck <bb...@gmail.com>, Damian Guy <da...@gmail.com>, Guozhang Wang <wa...@gmail.com>
Closes #3375 from mjsax/kafka-5474-eos-standby-task
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/914e42a2
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/914e42a2
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/914e42a2
Branch: refs/heads/trunk
Commit: 914e42a28254ef6a4818b3fcdc2197db6fbe8e0f
Parents: bc47e9d
Author: Matthias J. Sax <ma...@confluent.io>
Authored: Wed Jun 21 17:16:48 2017 -0700
Committer: Guozhang Wang <wa...@gmail.com>
Committed: Wed Jun 21 17:16:48 2017 -0700
----------------------------------------------------------------------
.../main/scala/kafka/serializer/Decoder.scala | 18 +++---
.../main/scala/kafka/serializer/Encoder.scala | 24 +++++++
.../processor/internals/StandbyTask.java | 7 ++-
.../processor/internals/StreamTaskTest.java | 66 ++++++++++++++++++++
.../kafka/streams/tests/EosTestDriver.java | 16 ++---
5 files changed, 114 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/914e42a2/core/src/main/scala/kafka/serializer/Decoder.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/serializer/Decoder.scala b/core/src/main/scala/kafka/serializer/Decoder.scala
index 5dfad26..9c1190b 100644
--- a/core/src/main/scala/kafka/serializer/Decoder.scala
+++ b/core/src/main/scala/kafka/serializer/Decoder.scala
@@ -54,17 +54,19 @@ class StringDecoder(props: VerifiableProperties = null) extends Decoder[String]
}
/**
- * The string decoder translates bytes into strings. It uses UTF8 by default but takes
- * an optional property serializer.encoding to control this.
+ * The long decoder translates bytes into longs.
*/
class LongDecoder(props: VerifiableProperties = null) extends Decoder[Long] {
- val encoding =
- if(props == null)
- "UTF8"
- else
- props.getString("serializer.encoding", "UTF8")
-
def fromBytes(bytes: Array[Byte]): Long = {
ByteBuffer.wrap(bytes).getLong
}
}
+
+/**
+ * The integer decoder translates bytes into integers.
+ */
+class IntegerDecoder(props: VerifiableProperties = null) extends Decoder[Integer] {
+ def fromBytes(bytes: Array[Byte]): Integer = {
+ ByteBuffer.wrap(bytes).getInt()
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/914e42a2/core/src/main/scala/kafka/serializer/Encoder.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/serializer/Encoder.scala b/core/src/main/scala/kafka/serializer/Encoder.scala
index 020e73c..00e2d94 100644
--- a/core/src/main/scala/kafka/serializer/Encoder.scala
+++ b/core/src/main/scala/kafka/serializer/Encoder.scala
@@ -17,6 +17,8 @@
package kafka.serializer
+import java.nio.ByteBuffer
+
import kafka.utils.VerifiableProperties
/**
@@ -56,3 +58,25 @@ class StringEncoder(props: VerifiableProperties = null) extends Encoder[String]
else
s.getBytes(encoding)
}
+
+/**
+ * The long encoder translates longs into bytes.
+ */
+class LongEncoder(props: VerifiableProperties = null) extends Encoder[Long] {
+ override def toBytes(l: Long): Array[Byte] =
+ if(l == null)
+ null
+ else
+ ByteBuffer.allocate(8).putLong(l).array()
+}
+
+/**
+ * The integer encoder translates integers into bytes.
+ */
+class IntegerEncoder(props: VerifiableProperties = null) extends Encoder[Integer] {
+ override def toBytes(i: Integer): Array[Byte] =
+ if(i == null)
+ null
+ else
+ ByteBuffer.allocate(4).putInt(i).array()
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/914e42a2/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
index b06b998..5998c2b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
@@ -91,8 +91,7 @@ public class StandbyTask extends AbstractTask {
@Override
public void commit() {
log.trace("{} Committing", logPrefix);
- stateMgr.flush();
- stateMgr.checkpoint(Collections.<TopicPartition, Long>emptyMap());
+ flushAndCheckpointState();
// reinitialize offset limits
updateOffsetLimits();
}
@@ -106,6 +105,10 @@ public class StandbyTask extends AbstractTask {
@Override
public void suspend() {
log.debug("{} Suspending", logPrefix);
+ flushAndCheckpointState();
+ }
+
+ private void flushAndCheckpointState() {
stateMgr.flush();
stateMgr.checkpoint(Collections.<TopicPartition, Long>emptyMap());
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/914e42a2/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index 7d5b5ca..130783d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -499,6 +499,72 @@ public class StreamTaskTest {
assertThat(checkpoint.read(), equalTo(Collections.singletonMap(partition, offset + 1)));
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void shouldNotCheckpointOffsetsOnCommitIfEosIsEnabled() throws Exception {
+ final Map<String, Object> properties = config.originals();
+ properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);
+ final StreamsConfig testConfig = new StreamsConfig(properties);
+
+ final String storeName = "test";
+ final String changelogTopic = ProcessorStateManager.storeChangelogTopic("appId", storeName);
+ final InMemoryKeyValueStore inMemoryStore = new InMemoryKeyValueStore(storeName, null, null) {
+ @Override
+ public void init(final ProcessorContext context, final StateStore root) {
+ context.register(root, true, null);
+ }
+
+ @Override
+ public boolean persistent() {
+ return true;
+ }
+ };
+ Map<String, SourceNode> sourceByTopics = new HashMap() {
+ {
+ put(partition1.topic(), source1);
+ put(partition2.topic(), source2);
+ }
+ };
+ final ProcessorTopology topology = new ProcessorTopology(Collections.<ProcessorNode>emptyList(),
+ sourceByTopics,
+ Collections.<String, SinkNode>emptyMap(),
+ Collections.<StateStore>singletonList(inMemoryStore),
+ Collections.singletonMap(storeName, changelogTopic),
+ Collections.<StateStore>emptyList());
+
+ final TopicPartition partition = new TopicPartition(changelogTopic, 0);
+
+ restoreStateConsumer.updatePartitions(changelogTopic,
+ Collections.singletonList(
+ new PartitionInfo(changelogTopic, 0, null, new Node[0], new Node[0])));
+ restoreStateConsumer.updateEndOffsets(Collections.singletonMap(partition, 0L));
+ restoreStateConsumer.updateBeginningOffsets(Collections.singletonMap(partition, 0L));
+
+ final long offset = 543L;
+ final StreamTask streamTask = new StreamTask(taskId00, "appId", partitions, topology, consumer,
+ changelogReader, testConfig, streamsMetrics, stateDirectory, null, time, producer) {
+
+ @Override
+ RecordCollector createRecordCollector() {
+ return new NoOpRecordCollector() {
+ @Override
+ public Map<TopicPartition, Long> offsets() {
+
+ return Collections.singletonMap(partition, offset);
+ }
+ };
+ }
+ };
+
+ time.sleep(testConfig.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG));
+
+ streamTask.commit();
+
+ final File checkpointFile = new File(stateDirectory.directoryForTask(taskId00),
+ ProcessorStateManager.CHECKPOINT_FILE_NAME);
+ assertFalse(checkpointFile.exists());
+ }
+
@Test
public void shouldThrowIllegalStateExceptionIfCurrentNodeIsNotNullWhenPunctuateCalled() throws Exception {
((ProcessorContextImpl) task.processorContext()).setCurrentNode(processor);
http://git-wip-us.apache.org/repos/asf/kafka/blob/914e42a2/streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java
index 07a0b1a..a1b813b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java
@@ -334,18 +334,19 @@ public class EosTestDriver extends SmokeTestUtil {
final String receivedKey = stringDeserializer.deserialize(receivedRecord.topic(), receivedRecord.key());
final int receivedValue = integerDeserializer.deserialize(receivedRecord.topic(), receivedRecord.value());
final String key = stringDeserializer.deserialize(input.topic(), input.key());
- final Integer value = integerDeserializer.deserialize(input.topic(), input.value());
+ final int value = integerDeserializer.deserialize(input.topic(), input.value());
Integer min = currentMinPerKey.get(key);
if (min == null) {
min = value;
+ } else {
+ min = Math.min(min, value);
}
- min = Math.min(min, value);
currentMinPerKey.put(key, min);
if (!receivedKey.equals(key) || receivedValue != min) {
- throw new RuntimeException("Result verification failed for " + receivedRecord + " expected <" + key + "," + value + "> but was <" + receivedKey + "," + receivedValue + ">");
+ throw new RuntimeException("Result verification failed for " + receivedRecord + " expected <" + key + "," + min + "> but was <" + receivedKey + "," + receivedValue + ">");
}
}
} catch (final NullPointerException e) {
@@ -374,17 +375,18 @@ public class EosTestDriver extends SmokeTestUtil {
final String receivedKey = stringDeserializer.deserialize(receivedRecord.topic(), receivedRecord.key());
final long receivedValue = longDeserializer.deserialize(receivedRecord.topic(), receivedRecord.value());
final String key = stringDeserializer.deserialize(input.topic(), input.key());
- final Integer value = integerDeserializer.deserialize(input.topic(), input.value());
+ final int value = integerDeserializer.deserialize(input.topic(), input.value());
Long sum = currentSumPerKey.get(key);
if (sum == null) {
- sum = 0L;
+ sum = (long) value;
+ } else {
+ sum += value;
}
- sum += value;
currentSumPerKey.put(key, sum);
if (!receivedKey.equals(key) || receivedValue != sum) {
- throw new RuntimeException("Result verification failed for " + receivedRecord + " expected <" + key + "," + value + "> but was <" + receivedKey + "," + receivedValue + ">");
+ throw new RuntimeException("Result verification failed for " + receivedRecord + " expected <" + key + "," + sum + "> but was <" + receivedKey + "," + receivedValue + ">");
}
}
} catch (final NullPointerException e) {