You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2021/12/21 21:35:40 UTC
[pinot] branch master updated: Fix realtime ingestion when an entire batch of messages is filtered out (#7927)
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new f8c7e1f Fix realtime ingestion when an entire batch of messages is filtered out (#7927)
f8c7e1f is described below
commit f8c7e1fc8603f4091e418f3841dcb6bc2d75d5d8
Author: Richard Startin <ri...@startree.ai>
AuthorDate: Tue Dec 21 21:35:17 2021 +0000
Fix realtime ingestion when an entire batch of messages is filtered out (#7927)
---
.../realtime/LLRealtimeSegmentDataManager.java | 25 ++++++-----
.../tests/BaseClusterIntegrationTest.java | 6 ++-
.../tests/ClusterIntegrationTestUtils.java | 10 ++++-
.../tests/LLCRealtimeClusterIntegrationTest.java | 5 +++
.../plugin/stream/kafka20/KafkaMessageBatch.java | 30 ++++++++++----
.../kafka20/KafkaPartitionLevelConsumer.java | 48 +++++++++-------------
.../kinesis/KinesisPartitionGroupOffset.java | 2 +-
.../stream/pulsar/MessageIdStreamOffset.java | 2 +-
.../org/apache/pinot/spi/stream/LongMsgOffset.java | 2 +-
.../org/apache/pinot/spi/stream/MessageBatch.java | 17 +++++++-
.../pinot/spi/stream/StreamPartitionMsgOffset.java | 2 +-
11 files changed, 91 insertions(+), 58 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
index c68efec..62c39a6 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
@@ -33,7 +33,6 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nullable;
@@ -85,7 +84,6 @@ import org.apache.pinot.spi.stream.StreamMessageDecoder;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffsetFactory;
-import org.apache.pinot.spi.stream.TransientConsumerException;
import org.apache.pinot.spi.utils.CommonConstants.ConsumerState;
import org.apache.pinot.spi.utils.CommonConstants.Segment.Realtime.CompletionMode;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
@@ -401,16 +399,12 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
.fetchMessages(_currentOffset, null, _partitionLevelStreamConfig.getFetchTimeoutMillis());
_endOfPartitionGroup = messageBatch.isEndOfPartitionGroup();
_consecutiveErrorCount = 0;
- } catch (TimeoutException e) {
- handleTransientStreamErrors(e);
- continue;
- } catch (TransientConsumerException e) {
- handleTransientStreamErrors(e);
- continue;
} catch (PermanentConsumerException e) {
_segmentLogger.warn("Permanent exception from stream when fetching messages, stopping consumption", e);
throw e;
} catch (Exception e) {
+ // all exceptions but PermanentConsumerException are handled the same way
+ // can be a TimeoutException or TransientConsumerException routinely
// Unknown exception from stream. Treat as a transient exception.
// One such exception seen so far is java.net.SocketTimeoutException
handleTransientStreamErrors(e);
@@ -423,12 +417,17 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
consecutiveIdleCount = 0;
// We consumed something. Update the highest stream offset as well as partition-consuming metric.
// TODO Issue 5359 Need to find a way to bump metrics without getting actual offset value.
-// _serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.HIGHEST_KAFKA_OFFSET_CONSUMED,
-// _currentOffset.getOffset());
-// _serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED,
-// _currentOffset.getOffset());
+ //_serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.HIGHEST_KAFKA_OFFSET_CONSUMED,
+ //_currentOffset.getOffset());
+ //_serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED,
+ //_currentOffset.getOffset());
_serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.LLC_PARTITION_CONSUMING, 1);
lastUpdatedOffset = _streamPartitionMsgOffsetFactory.create(_currentOffset);
+ } else if (messageBatch.getUnfilteredMessageCount() > 0) {
+ // we consumed something from the stream but filtered all the content out,
+ // so we need to advance the offsets to avoid getting stuck
+ _currentOffset = messageBatch.getOffsetOfNextBatch();
+ lastUpdatedOffset = _streamPartitionMsgOffsetFactory.create(_currentOffset);
} else {
// We did not consume any rows. Update the partition-consuming metric only if we have been idling for a long
// time.
@@ -559,7 +558,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
if (streamMessageCount != 0) {
_segmentLogger.debug("Indexed {} messages ({} messages read from stream) current offset {}", indexedMessageCount,
streamMessageCount, _currentOffset);
- } else {
+ } else if (messagesAndOffsets.getUnfilteredMessageCount() == 0) {
// If there were no messages to be fetched from stream, wait for a little bit as to avoid hammering the stream
Uninterruptibles.sleepUninterruptibly(idlePipeSleepTimeMillis, TimeUnit.MILLISECONDS);
}
diff --git a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
index ab5e028..b98c7c4 100644
--- a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
+++ b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
@@ -473,7 +473,11 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
ClusterIntegrationTestUtils
.pushAvroIntoKafka(avroFiles, "localhost:" + getKafkaPort(), getKafkaTopic(), getMaxNumKafkaMessagesPerBatch(),
- getKafkaMessageHeader(), getPartitionColumn());
+ getKafkaMessageHeader(), getPartitionColumn(), injectTombstones());
+ }
+
+ protected boolean injectTombstones() {
+ return false;
}
protected List<File> getAllAvroFiles()
diff --git a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterIntegrationTestUtils.java b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterIntegrationTestUtils.java
index 2b6231c..fc78bf4 100644
--- a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterIntegrationTestUtils.java
+++ b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterIntegrationTestUtils.java
@@ -317,7 +317,8 @@ public class ClusterIntegrationTestUtils {
* @throws Exception
*/
public static void pushAvroIntoKafka(List<File> avroFiles, String kafkaBroker, String kafkaTopic,
- int maxNumKafkaMessagesPerBatch, @Nullable byte[] header, @Nullable String partitionColumn)
+ int maxNumKafkaMessagesPerBatch, @Nullable byte[] header, @Nullable String partitionColumn,
+ boolean injectTombstones)
throws Exception {
Properties properties = new Properties();
properties.put("metadata.broker.list", kafkaBroker);
@@ -329,6 +330,13 @@ public class ClusterIntegrationTestUtils {
StreamDataProvider.getStreamDataProducer(KafkaStarterUtils.KAFKA_PRODUCER_CLASS_NAME, properties);
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream(65536)) {
+ if (injectTombstones) {
+ // publish lots of tombstones to livelock the consumer if it can't handle this properly
+ for (int i = 0; i < 1000; i++) {
+ // publish a tombstone first
+ producer.produce(kafkaTopic, Longs.toByteArray(System.currentTimeMillis()), null);
+ }
+ }
for (File avroFile : avroFiles) {
try (DataFileStream<GenericRecord> reader = AvroUtils.getAvroReader(avroFile)) {
BinaryEncoder binaryEncoder = new EncoderFactory().directBinaryEncoder(outputStream, null);
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
index 3e8dc6a..3c9f8ac 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
@@ -64,6 +64,11 @@ public class LLCRealtimeClusterIntegrationTest extends RealtimeClusterIntegratio
private final long _startTime = System.currentTimeMillis();
@Override
+ protected boolean injectTombstones() {
+ return true;
+ }
+
+ @Override
protected boolean useLlc() {
return true;
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaMessageBatch.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaMessageBatch.java
index db91da7..081334d 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaMessageBatch.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaMessageBatch.java
@@ -18,10 +18,7 @@
*/
package org.apache.pinot.plugin.stream.kafka20;
-import java.util.ArrayList;
import java.util.List;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.common.utils.Bytes;
import org.apache.pinot.plugin.stream.kafka.MessageAndOffset;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.MessageBatch;
@@ -30,12 +27,19 @@ import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
public class KafkaMessageBatch implements MessageBatch<byte[]> {
- private List<MessageAndOffset> _messageList = new ArrayList<>();
+ private final List<MessageAndOffset> _messageList;
+ private final int _unfilteredMessageCount;
+ private final long _lastOffset;
- public KafkaMessageBatch(Iterable<ConsumerRecord<String, Bytes>> iterable) {
- for (ConsumerRecord<String, Bytes> record : iterable) {
- _messageList.add(new MessageAndOffset(record.value().get(), record.offset()));
- }
+ /**
+ * @param unfilteredMessageCount how many messages were received from the topic before being filtered
+ * @param lastOffset the offset of the last message in the batch
+ * @param batch the messages, which may be smaller than {@see unfilteredMessageCount}
+ */
+ public KafkaMessageBatch(int unfilteredMessageCount, long lastOffset, List<MessageAndOffset> batch) {
+ _messageList = batch;
+ _lastOffset = lastOffset;
+ _unfilteredMessageCount = unfilteredMessageCount;
}
@Override
@@ -44,6 +48,11 @@ public class KafkaMessageBatch implements MessageBatch<byte[]> {
}
@Override
+ public int getUnfilteredMessageCount() {
+ return _unfilteredMessageCount;
+ }
+
+ @Override
public byte[] getMessageAtIndex(int index) {
return _messageList.get(index).getMessage().array();
}
@@ -67,4 +76,9 @@ public class KafkaMessageBatch implements MessageBatch<byte[]> {
public StreamPartitionMsgOffset getNextStreamParitionMsgOffsetAtIndex(int index) {
return new LongMsgOffset(_messageList.get(index).getNextOffset());
}
+
+ @Override
+ public StreamPartitionMsgOffset getOffsetOfNextBatch() {
+ return new LongMsgOffset(_lastOffset + 1);
+ }
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumer.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumer.java
index f816b81..68bbc9e 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumer.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumer.java
@@ -18,61 +18,51 @@
*/
package org.apache.pinot.plugin.stream.kafka20;
-import com.google.common.collect.Iterables;
-import java.io.IOException;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.TimeoutException;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.utils.Bytes;
+import org.apache.pinot.plugin.stream.kafka.MessageAndOffset;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.MessageBatch;
import org.apache.pinot.spi.stream.PartitionLevelConsumer;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHandler
implements PartitionLevelConsumer {
- private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPartitionLevelConsumer.class);
public KafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
}
@Override
- public MessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, StreamPartitionMsgOffset endMsgOffset,
- int timeoutMillis)
- throws TimeoutException {
+ public MessageBatch<byte[]> fetchMessages(StreamPartitionMsgOffset startMsgOffset,
+ StreamPartitionMsgOffset endMsgOffset, int timeoutMillis) {
final long startOffset = ((LongMsgOffset) startMsgOffset).getOffset();
final long endOffset = endMsgOffset == null ? Long.MAX_VALUE : ((LongMsgOffset) endMsgOffset).getOffset();
return fetchMessages(startOffset, endOffset, timeoutMillis);
}
- public MessageBatch fetchMessages(long startOffset, long endOffset, int timeoutMillis)
- throws TimeoutException {
+ public MessageBatch<byte[]> fetchMessages(long startOffset, long endOffset, int timeoutMillis) {
_consumer.seek(_topicPartition, startOffset);
ConsumerRecords<String, Bytes> consumerRecords = _consumer.poll(Duration.ofMillis(timeoutMillis));
- final Iterable<ConsumerRecord<String, Bytes>> messageAndOffsetIterable =
- buildOffsetFilteringIterable(consumerRecords.records(_topicPartition), startOffset, endOffset);
- return new KafkaMessageBatch(messageAndOffsetIterable);
- }
-
- private Iterable<ConsumerRecord<String, Bytes>> buildOffsetFilteringIterable(
- final List<ConsumerRecord<String, Bytes>> messageAndOffsets, final long startOffset, final long endOffset) {
- return Iterables.filter(messageAndOffsets, input -> {
- // Filter messages that are either null or have an offset ∉ [startOffset, endOffset]
- return input != null && input.value() != null && input.offset() >= startOffset && (endOffset > input.offset()
- || endOffset == -1);
- });
- }
-
- @Override
- public void close()
- throws IOException {
- super.close();
+ List<ConsumerRecord<String, Bytes>> messageAndOffsets = consumerRecords.records(_topicPartition);
+ List<MessageAndOffset> filtered = new ArrayList<>(messageAndOffsets.size());
+ long lastOffset = startOffset;
+ for (ConsumerRecord<String, Bytes> messageAndOffset : messageAndOffsets) {
+ Bytes message = messageAndOffset.value();
+ long offset = messageAndOffset.offset();
+ if (offset >= startOffset & (endOffset > offset | endOffset == -1)) {
+ if (message != null) {
+ filtered.add(new MessageAndOffset(message.get(), offset));
+ }
+ lastOffset = offset;
+ }
+ }
+ return new KafkaMessageBatch(messageAndOffsets.size(), lastOffset, filtered);
}
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisPartitionGroupOffset.java b/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisPartitionGroupOffset.java
index c2b06d7..4ee247f 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisPartitionGroupOffset.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisPartitionGroupOffset.java
@@ -77,7 +77,7 @@ public class KinesisPartitionGroupOffset implements StreamPartitionMsgOffset {
}
@Override
- public int compareTo(Object o) {
+ public int compareTo(StreamPartitionMsgOffset o) {
Preconditions.checkNotNull(o);
KinesisPartitionGroupOffset other = (KinesisPartitionGroupOffset) o;
Preconditions.checkNotNull(other._shardToStartSequenceMap);
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/MessageIdStreamOffset.java b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/MessageIdStreamOffset.java
index 45cfca0..cf23580 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/MessageIdStreamOffset.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/MessageIdStreamOffset.java
@@ -60,7 +60,7 @@ public class MessageIdStreamOffset implements StreamPartitionMsgOffset {
}
@Override
- public int compareTo(Object other) {
+ public int compareTo(StreamPartitionMsgOffset other) {
MessageIdStreamOffset messageIdStreamOffset = (MessageIdStreamOffset) other;
return _messageId.compareTo(messageIdStreamOffset.getMessageId());
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/LongMsgOffset.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/LongMsgOffset.java
index 5ee78ec..eb1d4f5 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/LongMsgOffset.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/LongMsgOffset.java
@@ -42,7 +42,7 @@ public class LongMsgOffset implements StreamPartitionMsgOffset {
}
@Override
- public int compareTo(Object other) {
+ public int compareTo(StreamPartitionMsgOffset other) {
return Long.compare(_offset, ((LongMsgOffset) other)._offset);
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java
index 02c721f..dfcf369 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java
@@ -31,12 +31,18 @@ import org.apache.pinot.spi.annotations.InterfaceStability;
@InterfaceStability.Stable
public interface MessageBatch<T> {
/**
- *
- * @return number of messages returned from the stream
+ * @return number of available messages
*/
int getMessageCount();
/**
+ * @return number of messages returned from the stream
+ */
+ default int getUnfilteredMessageCount() {
+ return getMessageCount();
+ }
+
+ /**
* Returns the message at a particular index inside a set of messages returned from the stream.
* @param index
* @return
@@ -83,6 +89,13 @@ public interface MessageBatch<T> {
}
/**
+ * @return last offset in the batch
+ */
+ default StreamPartitionMsgOffset getOffsetOfNextBatch() {
+ return getNextStreamParitionMsgOffsetAtIndex(getMessageCount() - 1);
+ }
+
+ /**
* Returns true if end of the consumer detects that no more records can be read from this partition group for good
*/
default boolean isEndOfPartitionGroup() {
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamPartitionMsgOffset.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamPartitionMsgOffset.java
index 7a5018c..6315deb 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamPartitionMsgOffset.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamPartitionMsgOffset.java
@@ -40,7 +40,7 @@ import org.apache.pinot.spi.annotations.InterfaceStability;
* versions of the stream implementation
*/
@InterfaceStability.Evolving
-public interface StreamPartitionMsgOffset extends Comparable {
+public interface StreamPartitionMsgOffset extends Comparable<StreamPartitionMsgOffset> {
/**
* A serialized representation of the offset object as a String.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org