Skip to content

STORM-2607 1.x #2369

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
asfgit merged 2 commits into from
Oct 17, 2017
Merged

STORM-2607 1.x #2369

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4fc47b2
[STORM-2607] Offset consumer + 1
Jun 29, 2017
148ee66
STORM-2607: Switch OffsetManager to track earliest uncommitted offset…
srdo Jul 1, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,7 +202,7 @@ private void initialize(Collection<TopicPartition> partitions) {
}

/**
* sets the cursor to the location dictated by the first poll strategy and returns the fetch offset
* Sets the cursor to the location dictated by the first poll strategy and returns the fetch offset.
*/
private long doSeek(TopicPartition tp, OffsetAndMetadata committedOffset) {
if (committedOffset != null) { // offset was committed for this TopicPartition
Expand All @@ -211,8 +211,8 @@ private long doSeek(TopicPartition tp, OffsetAndMetadata committedOffset) {
} else if (firstPollOffsetStrategy.equals(LATEST)) {
kafkaConsumer.seekToEnd(Collections.singleton(tp));
} else {
// By default polling starts at the last committed offset. +1 to point fetch to the first uncommitted offset.
kafkaConsumer.seek(tp, committedOffset.offset() + 1);
// By default polling starts at the last committed offset, i.e. the first offset that was not marked as processed.
kafkaConsumer.seek(tp, committedOffset.offset());
}
} else { // no commits have ever been done, so start at the beginning or end depending on the strategy
if (firstPollOffsetStrategy.equals(EARLIEST) || firstPollOffsetStrategy.equals(UNCOMMITTED_EARLIEST)) {
Expand Down
80 changes: 48 additions & 32 deletions ...storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ public class OffsetManager {
/* First offset to be fetched. It is either set to the beginning, end, or to the first uncommitted offset.
* Initial value depends on offset strategy. See KafkaSpoutConsumerRebalanceListener */
private final long initialFetchOffset;
// Last offset committed to Kafka. Initially it is set to fetchOffset - 1
// Committed offset, i.e. the offset where processing will resume if the spout restarts. Initially it is set to fetchOffset.
private long committedOffset;
// Emitted Offsets List
private final NavigableSet<Long> emittedOffsets = new TreeSet<>();
Expand All @@ -47,8 +47,8 @@ public class OffsetManager {
public OffsetManager(TopicPartition tp, long initialFetchOffset) {
this.tp = tp;
this.initialFetchOffset = initialFetchOffset;
this.committedOffset = initialFetchOffset - 1;
LOG.debug("Instantiated {}", this);
this.committedOffset = initialFetchOffset;
LOG.debug("Instantiated {}", this.toString());
}

public void addToAckMsgs(KafkaSpoutMessageId msgId) { // O(Log N)
Expand All @@ -60,9 +60,11 @@ public void addToEmitMsgs(long offset) {
}

/**
* An offset is only committed when all records with lower offset have been
* An offset can only be committed when all emitted records with lower offset have been
* acked. This guarantees that all offsets smaller than the committedOffset
* have been delivered.
* have been delivered, or that those offsets no longer exist in Kafka.
* <p/>
* The returned offset points to the earliest uncommitted offset, and matches the semantics of the KafkaConsumer.commitSync API.
*
* @return the next OffsetAndMetadata to commit, or null if no offset is
* ready to commit.
Expand All @@ -75,66 +77,74 @@ public OffsetAndMetadata findNextCommitOffset() {

for (KafkaSpoutMessageId currAckedMsg : ackedMsgs) { // complexity is that of a linear scan on a TreeMap
currOffset = currAckedMsg.offset();
if (currOffset == nextCommitOffset + 1) { // found the next offset to commit
if (currOffset == nextCommitOffset) { // found the next offset to commit
found = true;
nextCommitMsg = currAckedMsg;
nextCommitOffset = currOffset;
} else if (currOffset > nextCommitOffset + 1) {
if (emittedOffsets.contains(nextCommitOffset + 1)) {
LOG.debug("topic-partition [{}] has non-continuous offset [{}]. It will be processed in a subsequent batch.", tp, currOffset);
nextCommitOffset = currOffset + 1;
} else if (currOffset > nextCommitOffset) {
if (emittedOffsets.contains(nextCommitOffset)) {
LOG.debug("topic-partition [{}] has non-sequential offset [{}]."
+ " It will be processed in a subsequent batch.", tp, currOffset);
break;
} else {
/*
This case will arise in case of non contiguous offset being processed.
So, if the topic doesn't contain offset = committedOffset + 1 (possible
This case will arise in case of non-sequential offset being processed.
So, if the topic doesn't contain offset = nextCommitOffset (possible
if the topic is compacted or deleted), the consumer should jump to
the next logical point in the topic. Next logical offset should be the
first element after committedOffset in the ascending ordered emitted set.
first element after nextCommitOffset in the ascending ordered emitted set.
*/
LOG.debug("Processed non contiguous offset. (committedOffset+1) is no longer part of the topic. Committed: [{}], Processed: [{}]", committedOffset, currOffset);
final Long nextEmittedOffset = emittedOffsets.ceiling(nextCommitOffset + 1);
LOG.debug("Processed non-sequential offset."
+ " The earliest uncommitted offset is no longer part of the topic."
+ " Missing offset: [{}], Processed: [{}]", nextCommitOffset, currOffset);
final Long nextEmittedOffset = emittedOffsets.ceiling(nextCommitOffset);
if (nextEmittedOffset != null && currOffset == nextEmittedOffset) {
found = true;
LOG.debug("Found committable offset: [{}] after missing offset: [{}], skipping to the committable offset",
currOffset, nextCommitOffset);
nextCommitMsg = currAckedMsg;
nextCommitOffset = currOffset;
nextCommitOffset = currOffset + 1;
} else {
LOG.debug("topic-partition [{}] has non-continuous offset [{}]. Next Offset to commit should be [{}]", tp, currOffset, nextEmittedOffset);
LOG.debug("Topic-partition [{}] has non-sequential offset [{}]."
+ " Next offset to commit should be [{}]", tp, currOffset, nextCommitOffset);
break;
}
}
} else {
throw new IllegalStateException("The offset [" + currOffset + "] is below the current committed "
+ "offset [" + committedOffset + "] for [" + tp + "]."
throw new IllegalStateException("The offset [" + currOffset + "] is below the current nextCommitOffset "
+ "[" + nextCommitOffset + "] for [" + tp + "]."
+ " This should not be possible, and likely indicates a bug in the spout's acking or emit logic.");
}
}

OffsetAndMetadata nextCommitOffsetAndMetadata = null;
if (found) {
nextCommitOffsetAndMetadata = new OffsetAndMetadata(nextCommitOffset, nextCommitMsg.getMetadata(Thread.currentThread()));
LOG.debug("topic-partition [{}] has offsets [{}-{}] ready to be committed", tp, committedOffset + 1, nextCommitOffsetAndMetadata.offset());
nextCommitOffsetAndMetadata = new OffsetAndMetadata(nextCommitOffset,
nextCommitMsg.getMetadata(Thread.currentThread()));
LOG.debug("Topic-partition [{}] has offsets [{}-{}] ready to be committed."
+ " Processing will resume at offset [{}] if the spout restarts",
tp, committedOffset, nextCommitOffsetAndMetadata.offset() - 1, nextCommitOffsetAndMetadata.offset());
} else {
LOG.debug("topic-partition [{}] has NO offsets ready to be committed", tp);
LOG.debug("Topic-partition [{}] has no offsets ready to be committed", tp);
}
LOG.trace("{}", this);
return nextCommitOffsetAndMetadata;
}

/**
* Marks an offset has committed. This method has side effects - it sets the
* Marks an offset as committed. This method has side effects - it sets the
* internal state in such a way that future calls to
* {@link #findNextCommitOffset()} will return offsets greater than the
* {@link #findNextCommitOffset()} will return offsets greater than or equal to the
* offset specified, if any.
*
* @param committedOffset offset to be marked as committed
* @param committedOffset The committed offset. All lower offsets are expected to have been committed.
* @return Number of offsets committed in this commit
*/
public long commit(OffsetAndMetadata committedOffset) {
final long preCommitCommittedOffsets = this.committedOffset;
final long preCommitCommittedOffset = this.committedOffset;
long numCommittedOffsets = 0;
this.committedOffset = committedOffset.offset();
for (Iterator<KafkaSpoutMessageId> iterator = ackedMsgs.iterator(); iterator.hasNext();) {
if (iterator.next().offset() <= committedOffset.offset()) {
if (iterator.next().offset() < committedOffset.offset()) {
iterator.remove();
numCommittedOffsets++;
} else {
Expand All @@ -143,7 +153,7 @@ public long commit(OffsetAndMetadata committedOffset) {
}

for (Iterator<Long> iterator = emittedOffsets.iterator(); iterator.hasNext();) {
if (iterator.next() <= committedOffset.offset()) {
if (iterator.next() < committedOffset.offset()) {
iterator.remove();
} else {
break;
Expand All @@ -152,8 +162,9 @@ public long commit(OffsetAndMetadata committedOffset) {

LOG.trace("{}", this);

LOG.debug("Committed [{}] offsets in the range [{}-{}] for topic-partition [{}].",
numCommittedOffsets, preCommitCommittedOffsets + 1, this.committedOffset, tp);
LOG.debug("Committed [{}] offsets in the range [{}-{}] for topic-partition [{}]."
+ " Processing will resume at [{}] if the spout restarts.",
numCommittedOffsets, preCommitCommittedOffset, this.committedOffset - 1, tp, this.committedOffset);

return numCommittedOffsets;
}
Expand All @@ -173,9 +184,14 @@ public boolean contains(ConsumerRecord record) {
public boolean contains(KafkaSpoutMessageId msgId) {
return ackedMsgs.contains(msgId);
}

//VisibleForTesting
boolean containsEmitted(long offset) {
return emittedOffsets.contains(offset);
}

@Override
public String toString() {
public final String toString() {
return "OffsetManager{"
+ "topic-partition=" + tp
+ ", fetchOffset=" + initialFetchOffset
Expand Down
6 changes: 2 additions & 4 deletions ...l/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutCommitTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,8 +42,6 @@

import static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder;

import org.mockito.stubbing.OngoingStubbing;

public class KafkaSpoutCommitTest {

private final long offsetCommitPeriodMs = 2_000;
Expand Down Expand Up @@ -107,10 +105,10 @@ public void testCommitSuccessWithOffsetVoids() {
inOrder.verify(consumerMock).commitSync(commitCapture.capture());
inOrder.verify(consumerMock).poll(anyLong());

//verify that Offset 9 was last committed offset
//verify that Offset 10 was last committed offset, since this is the offset the spout should resume at
Map<TopicPartition, OffsetAndMetadata> commits = commitCapture.getValue();
assertTrue(commits.containsKey(partition));
assertEquals(9, commits.get(partition).offset());
assertEquals(10, commits.get(partition).offset());
}
}

Expand Down
4 changes: 2 additions & 2 deletions ...torm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRebalanceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@
import org.mockito.MockitoAnnotations;

import static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder;
import static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder;
import static org.mockito.Matchers.anyObject;
import static org.mockito.Matchers.eq;

import java.util.HashSet;
Expand Down Expand Up @@ -247,6 +247,6 @@ public void testReassignPartitionSeeksForOnlyNewPartitions() {
//This partition was previously assigned, so the consumer position shouldn't change
verify(consumerMock, never()).seek(eq(assignedPartition), anyLong());
//This partition is new, and should start at the committed offset
verify(consumerMock).seek(newPartition, committedOffset + 1);
verify(consumerMock).seek(newPartition, committedOffset);
}
}
8 changes: 5 additions & 3 deletions ...orm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRetryLimitTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.*;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
Expand Down Expand Up @@ -45,6 +44,9 @@
import org.mockito.MockitoAnnotations;

import static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder;
import static org.mockito.Matchers.anyList;
import static org.mockito.Matchers.anyLong;
import static org.mockito.Matchers.anyString;

public class KafkaSpoutRetryLimitTest {

Expand Down Expand Up @@ -108,9 +110,9 @@ public void testFailingTupleCompletesAckAfterRetryLimitIsMet() {
inOrder.verify(consumerMock).commitSync(commitCapture.capture());
inOrder.verify(consumerMock).poll(anyLong());

//verify that Offset 3 was committed for the given TopicPartition
//verify that offset 4 was committed for the given TopicPartition, since processing should resume at 4.
assertTrue(commitCapture.getValue().containsKey(partition));
assertEquals(lastOffset, ((OffsetAndMetadata) (commitCapture.getValue().get(partition))).offset());
assertEquals(lastOffset + 1, ((OffsetAndMetadata) (commitCapture.getValue().get(partition))).offset());
}
}

Expand Down
4 changes: 2 additions & 2 deletions ...rm-kafka-client/src/test/java/org/apache/storm/kafka/spout/SingleTopicKafkaSpoutTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ private void verifyAllMessagesCommitted(long messageCount) {
Map<TopicPartition, OffsetAndMetadata> commits = commitCapture.getValue();
assertThat("Expected commits for only one topic partition", commits.entrySet().size(), is(1));
OffsetAndMetadata offset = commits.entrySet().iterator().next().getValue();
assertThat("Expected committed offset to cover all emitted messages", offset.offset(), is(messageCount - 1));
assertThat("Expected committed offset to cover all emitted messages", offset.offset(), is(messageCount));
}

@Test
Expand Down Expand Up @@ -159,7 +159,7 @@ public void testSeekToCommittedOffsetIfConsumerPositionIsBehindWhenCommitting()
Map<TopicPartition, OffsetAndMetadata> capturedCommit = commitCapture.getValue();
TopicPartition expectedTp = new TopicPartition(SingleTopicKafkaSpoutConfiguration.TOPIC, 0);
assertThat("Should have committed to the right topic", capturedCommit, Matchers.hasKey(expectedTp));
assertThat("Should have committed all the acked messages", capturedCommit.get(expectedTp).offset(), is((long)messageCount - 1));
assertThat("Should have committed all the acked messages", capturedCommit.get(expectedTp).offset(), is((long)messageCount));

/* Verify that the following acked (now committed) tuples are not emitted again
* Since the consumer position was somewhere in the middle of the acked tuples when the commit happened,
Expand Down
Loading