Skip to content

STORM-2607: Storm-kafka-client never commits the last message on a partition. #2367

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 14, 2017
Merged

STORM-2607: Storm-kafka-client never commits the last message on a partition. #2367

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
14e98e7
[STORM-2607] Offset consumer + 1
Jun 29, 2017
d7bdc2d
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 @@ -197,7 +197,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 @@ -206,8 +206,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
83 changes: 48 additions & 35 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 @@ -16,6 +16,7 @@

package org.apache.storm.kafka.spout.internal;

import com.google.common.annotations.VisibleForTesting;
import java.util.Comparator;
import java.util.Iterator;
import java.util.NavigableSet;
Expand All @@ -38,7 +39,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 @@ -53,8 +54,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());

@hmcl hmcl Oct 11, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

toString() is redundant

@srdo srdo Oct 11, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, but I get a warning about leaking this from the constructor, which this fixes.

@hmcl hmcl Oct 11, 2017 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is not a big deal, but can you tell me what kinda warning you are talking about? I never heard of such a thing. When you concatenate a string with an object reference, the toString() method of that object reference is called. I am not sure why both approaches are any different.

@srdo srdo Oct 11, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The warning I get is just "leaking this in constructor", Netbeans has the following description: "Using this as parameter can be dangerous in the contructor because the object is not fully initialized.". IntelliJ and Eclipse have similar warnings I believe.

It's basically saying not to use this as a parameter from the constructor, because the called method may use fields that aren't initialized. I agree that leaking this from the constructor is not dangerous here, because sfl4j just calls toString, but I'd still like to get rid of the warning. The reason this is flagged is because this is used as a parameter to LOG.debug, and not as part of a string concatenation. The IDE can't tell whether this is used unsafely in LOG.debug, so it's flagged as a warning.

@hmcl hmcl Oct 11, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the explanation. I just noticed that this is called within its own constructor, and therefore makes sense. However, IntelliJ does not give me this warning.

@srdo srdo Oct 11, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's not enabled by default in IntelliJ. If you go to settings -> editor -> inspections and search for "this", there's one called "'this' reference escaped in object construction"

}

public void addToAckMsgs(KafkaSpoutMessageId msgId) { // O(Log N)
Expand All @@ -66,9 +67,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 @@ -80,69 +83,73 @@ 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
nextCommitMsg = currAckedMsg;
nextCommitOffset = currOffset;
} else if (currOffset > nextCommitOffset + 1) {
if (emittedOffsets.contains(nextCommitOffset + 1)) {
LOG.debug("topic-partition [{}] has non-continuous offset [{}]."
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 nextCommitOffsset 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) {
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 (nextCommitMsg != null) {
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 [{}] 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 @@ -151,7 +158,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 @@ -160,8 +167,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 offset [{}] if the spout restarts.",
numCommittedOffsets, preCommitCommittedOffset, this.committedOffset - 1, tp, this.committedOffset);

return numCommittedOffsets;
}
Expand All @@ -181,9 +189,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
9 changes: 5 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 @@ -39,9 +39,10 @@
import org.mockito.MockitoAnnotations;

import static org.apache.storm.kafka.spout.config.builder.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder;
import static org.mockito.ArgumentMatchers.anyList;

import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyObject;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
Expand Down Expand Up @@ -94,7 +95,7 @@ public void testCommitSuccessWithOffsetVoids() {
}

ArgumentCaptor<KafkaSpoutMessageId> messageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
verify(collectorMock, times(recordsForPartition.size())).emit(anyObject(), anyObject(), messageIds.capture());
verify(collectorMock, times(recordsForPartition.size())).emit(anyString(), anyList(), messageIds.capture());

for (KafkaSpoutMessageId messageId : messageIds.getAllValues()) {
spout.ack(messageId);
Expand All @@ -110,10 +111,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
21 changes: 12 additions & 9 deletions ...nal/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutEmitTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,6 @@
*/
package org.apache.storm.kafka.spout;

import static org.mockito.Matchers.anyLong;
import static org.mockito.Matchers.anyObject;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
Expand Down Expand Up @@ -45,6 +43,11 @@
import org.mockito.InOrder;

import static org.apache.storm.kafka.spout.config.builder.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyObject;
import static org.mockito.ArgumentMatchers.anyString;

public class KafkaSpoutEmitTest {

Expand Down Expand Up @@ -81,7 +84,7 @@ public void testNextTupleEmitsAtMostOneTuple() {

spout.nextTuple();

verify(collectorMock, times(1)).emit(anyObject(), anyObject(), anyObject());
verify(collectorMock, times(1)).emit(anyString(), anyList(), any(KafkaSpoutMessageId.class));
}

@Test
Expand All @@ -107,7 +110,7 @@ public void testNextTupleEmitsFailedMessagesEvenWhenMaxUncommittedOffsetsIsExcee
}

ArgumentCaptor<KafkaSpoutMessageId> messageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
verify(collectorMock, times(recordsForPartition.size())).emit(anyObject(), anyObject(), messageIds.capture());
verify(collectorMock, times(recordsForPartition.size())).emit(anyString(), anyList(), messageIds.capture());

for (KafkaSpoutMessageId messageId : messageIds.getAllValues()) {
spout.fail(messageId);
Expand All @@ -122,7 +125,7 @@ public void testNextTupleEmitsFailedMessagesEvenWhenMaxUncommittedOffsetsIsExcee
}

ArgumentCaptor<KafkaSpoutMessageId> retryMessageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
verify(collectorMock, times(recordsForPartition.size())).emit(anyObject(), anyObject(), retryMessageIds.capture());
verify(collectorMock, times(recordsForPartition.size())).emit(anyString(), anyList(), retryMessageIds.capture());

//Verify that the poll started at the earliest retriable tuple offset
List<Long> failedOffsets = new ArrayList<>();
Expand Down Expand Up @@ -182,7 +185,7 @@ public void testNextTupleEmitsAtMostMaxUncommittedOffsetsPlusMaxPollRecordsWhenR
}

ArgumentCaptor<KafkaSpoutMessageId> messageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
verify(collectorMock, times(firstPollRecordsForPartition.size())).emit(anyObject(), anyObject(), messageIds.capture());
verify(collectorMock, times(firstPollRecordsForPartition.size())).emit(anyString(), anyList(), messageIds.capture());

KafkaSpoutMessageId failedMessageId = messageIds.getAllValues().get(messageIds.getAllValues().size() - 1);
spout.fail(failedMessageId);
Expand All @@ -197,7 +200,7 @@ public void testNextTupleEmitsAtMostMaxUncommittedOffsetsPlusMaxPollRecordsWhenR
}

ArgumentCaptor<KafkaSpoutMessageId> retryBatchMessageIdsCaptor = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
verify(collectorMock, times(maxPollRecords)).emit(anyObject(), anyObject(), retryBatchMessageIdsCaptor.capture());
verify(collectorMock, times(maxPollRecords)).emit(anyString(), anyList(), retryBatchMessageIdsCaptor.capture());
reset(collectorMock);

//Check that the consumer started polling at the failed tuple offset
Expand All @@ -215,12 +218,12 @@ public void testNextTupleEmitsAtMostMaxUncommittedOffsetsPlusMaxPollRecordsWhenR
for (int i = 0; i < firstPollRecordsForPartition.size() + maxPollRecords; i++) {
spout.nextTuple();
}
verify(collectorMock, never()).emit(anyObject(), anyObject(), anyObject());
verify(collectorMock, never()).emit(any(), any(), any());

//Fail the last tuple, which brings the number of nonretriable tuples back under the limit, and check that the spout polls again
spout.fail(firstTupleFromRetryBatch);
spout.nextTuple();
verify(collectorMock, times(1)).emit(anyObject(), anyObject(), anyObject());
verify(collectorMock, times(1)).emit(anyString(), anyList(), any(KafkaSpoutMessageId.class));
}
}

Expand Down
8 changes: 4 additions & 4 deletions ...a-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutMessagingGuaranteeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,10 @@
import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.junit.Assert.assertThat;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyList;
import static org.mockito.Matchers.anyLong;
import static org.mockito.Matchers.eq;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
Expand Down
Loading