diff --git a/docs/reference-metrics.md b/docs/reference-metrics.md index 8a83a0b93e..b19cfd6188 100644 --- a/docs/reference-metrics.md +++ b/docs/reference-metrics.md @@ -72,6 +72,7 @@ The KoP metrics are exposed under "/metrics" at port `8000` along with Pulsar me | kop_server_MESSAGE_OUT | Counter | The consumer message out stats.
Available labels: *topic*, *partition*, *group*.
| | kop_server_ENTRIES_OUT | Counter | The consumer entries out stats.
Available labels: *topic*, *partition*, *group*.
| | kop_server_CONSUME_MESSAGE_CONVERSIONS | Counter | The consumer message conversions in stats.
Available labels: *topic*, *partition*.
| +| kop_server_WAITING_FETCHES_TRIGGERED | Counter | Number of fetches that have been delayed due to not enough data, and that have been unblocked because some message has been produced| ### Kop event metrics diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/DelayedFetch.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/DelayedFetch.java index 105c2644a0..811b36de2b 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/DelayedFetch.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/DelayedFetch.java @@ -15,36 +15,67 @@ import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperation; import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import lombok.extern.slf4j.Slf4j; +@Slf4j public class DelayedFetch extends DelayedOperation { private final Runnable callback; private final AtomicLong bytesReadable; private final int minBytes; + private final MessageFetchContext messageFetchContext; + private final AtomicBoolean restarted = new AtomicBoolean(); + private final AtomicBoolean someMessageProduced = new AtomicBoolean(); - protected DelayedFetch(long delayMs, AtomicLong bytesReadable, int minBytes, Runnable callback) { + protected DelayedFetch(long delayMs, AtomicLong bytesReadable, int minBytes, + MessageFetchContext messageFetchContext) { super(delayMs, Optional.empty()); - this.callback = callback; this.bytesReadable = bytesReadable; this.minBytes = minBytes; + this.messageFetchContext = messageFetchContext; + this.callback = messageFetchContext::complete; } @Override public void onExpiration() { + if (restarted.get()) { + return; + } callback.run(); } @Override public void onComplete() { + if (restarted.get()) { + return; + } callback.run(); } @Override public boolean tryComplete() { + if (someMessageProduced.get()) { + // if we are here then we were waiting for the condition + // someone wrote some messages to one of the topics + // trigger the Fetch from scratch + restarted.set(true); + messageFetchContext.onDataWrittenToSomePartition(); + return true; + } if (bytesReadable.get() < minBytes){ return false; } callback.run(); return true; } + + @Override + public boolean wakeup() { + // In the future we could notify the MessageFetchContext that the + // new data is only on this partition and not + // on other partitions + someMessageProduced.set(true); + return true; + } } diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index ed51012167..4b46613d50 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -56,6 +56,7 @@ import io.streamnative.pulsar.handlers.kop.utils.OffsetFinder; import io.streamnative.pulsar.handlers.kop.utils.TopicNameUtils; import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperation; +import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperationKey; import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperationPurgatory; import java.net.InetSocketAddress; import java.nio.ByteBuffer; @@ -941,6 +942,11 @@ protected void handleProduceRequest(KafkaHeaderAndRequest produceHar, mergedResponse.putAll(unauthorizedTopicResponsesMap); mergedResponse.putAll(invalidRequestResponses); resultFuture.complete(new ProduceResponse(mergedResponse)); + mergedResponse.forEach((_topicPartition, _response) -> { + if (_response.error == Errors.NONE) { + notifyPendingFetches(_topicPartition); + } + }); }); } }; @@ -979,6 +985,20 @@ protected void handleProduceRequest(KafkaHeaderAndRequest produceHar, } + private void notifyPendingFetches(TopicPartition topicPartition) { + ctx.executor().execute(() -> { + DelayedOperationKey.TopicPartitionOperationKey key = + new DelayedOperationKey.TopicPartitionOperationKey(topicPartition); + int matches = fetchPurgatory.checkAndComplete(key); + if (matches > 0) { + requestStats.getWaitingFetchesTriggered().add(matches); + if (log.isDebugEnabled()) { + log.debug("{} DelayedFetch woke up for {}", matches, topicPartition); + } + } + }); + } + private void validateRecords(short version, MemoryRecords records) { if (version >= 3) { Iterator iterator = records.batches().iterator(); diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java index 65c3cea99f..8e8324d1d6 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java @@ -75,6 +75,7 @@ public interface KopServerStats { String PREPARE_METADATA = "PREPARE_METADATA"; String MESSAGE_READ = "MESSAGE_READ"; String FETCH_DECODE = "FETCH_DECODE"; + String WAITING_FETCHES_TRIGGERED = "WAITING_FETCHES_TRIGGERED"; /** * Consumer stats. diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/MessageFetchContext.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/MessageFetchContext.java index a79b836d42..d1f974c6a3 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/MessageFetchContext.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/MessageFetchContext.java @@ -68,6 +68,7 @@ import org.apache.kafka.common.requests.IsolationLevel; import org.apache.kafka.common.requests.RequestHeader; import org.apache.kafka.common.requests.ResponseCallbackWrapper; +import org.apache.kafka.common.utils.SystemTime; import org.apache.pulsar.metadata.api.GetResult; /** @@ -83,6 +84,7 @@ protected MessageFetchContext newObject(Handle handle) { }; private final Handle recyclerHandle; + private long startTime; private Map> responseData; private ConcurrentLinkedQueue decodeResults; private KafkaRequestHandler requestHandler; @@ -95,7 +97,7 @@ protected MessageFetchContext newObject(Handle handle) { private RequestHeader header; private volatile CompletableFuture resultFuture; private AtomicBoolean hasComplete; - private AtomicLong bytesReadable; + private AtomicLong bytesRead; private DelayedOperationPurgatory fetchPurgatory; private String namespacePrefix; @@ -119,8 +121,9 @@ public static MessageFetchContext get(KafkaRequestHandler requestHandler, context.header = kafkaHeaderAndRequest.getHeader(); context.resultFuture = resultFuture; context.hasComplete = new AtomicBoolean(false); - context.bytesReadable = new AtomicLong(0); + context.bytesRead = new AtomicLong(0); context.fetchPurgatory = fetchPurgatory; + context.startTime = SystemTime.SYSTEM.hiResClockMs(); return context; } @@ -142,6 +145,7 @@ public static MessageFetchContext getForTest(FetchRequest fetchRequest, context.header = null; context.resultFuture = resultFuture; context.hasComplete = new AtomicBoolean(false); + context.startTime = SystemTime.SYSTEM.hiResClockMs(); return context; } @@ -163,7 +167,7 @@ private void recycle() { header = null; resultFuture = null; hasComplete = null; - bytesReadable = null; + bytesRead = null; fetchPurgatory = null; namespacePrefix = null; recyclerHandle.recycle(this); @@ -195,15 +199,47 @@ private void addErrorPartitionResponse(TopicPartition topicPartition, Errors err private void tryComplete() { if (resultFuture != null && responseData.size() >= fetchRequest.fetchData().size() && hasComplete.compareAndSet(false, true)) { - DelayedFetch delayedFetch = new DelayedFetch(fetchRequest.maxWait(), bytesReadable, - fetchRequest.minBytes(), this::complete); - List delayedFetchKeys = - fetchRequest.fetchData().keySet().stream() - .map(DelayedOperationKey.TopicPartitionOperationKey::new).collect(Collectors.toList()); - fetchPurgatory.tryCompleteElseWatch(delayedFetch, delayedFetchKeys); + boolean errorsOccurred = false; + if (responseData + .values() + .stream() + .anyMatch(p->p.error != Errors.NONE)) { + // if there is an error no need to wait, the fetch must fail + // as soon as possible + errorsOccurred = true; + } + long now = SystemTime.SYSTEM.hiResClockMs(); + long currentWait = now - this.startTime; + long remainingMaxWait = fetchRequest.maxWait() - currentWait; + long maxWait = Math.min(remainingMaxWait, fetchRequest.maxWait()); + if (bytesRead.get() < fetchRequest.minBytes() && !errorsOccurred && maxWait > 0) { + // we haven't read enough data, need to wait + DelayedFetch delayedFetch = new DelayedFetch(maxWait, bytesRead, + fetchRequest.minBytes(), this); + List delayedFetchKeys = + fetchRequest.fetchData().keySet().stream() + .map(DelayedOperationKey.TopicPartitionOperationKey::new).collect(Collectors.toList()); + fetchPurgatory.tryCompleteElseWatch(delayedFetch, delayedFetchKeys); + } else { + this.complete(); + } } } + /** + * Restart this Fetch, we were waiting for some data (minBytes) + * and someone wrote something on any of the watched partitions. + */ + public void onDataWrittenToSomePartition() { + decodeResults.forEach(DecodeResult::recycle); + decodeResults.clear(); + bytesRead.set(0); + hasComplete.set(false); + responseData.clear(); + handleFetch(); + } + + public void complete() { if (resultFuture == null) { // the context has been recycled @@ -329,6 +365,9 @@ private void handlePartitionData(final TopicPartition topicPartition, // the future that is returned by getTopicConsumerManager is always completed normally topicManager.getTopicConsumerManager(fullTopicName).thenAccept(tcm -> { if (tcm == null) { + if (log.isDebugEnabled()) { + log.debug("Fetch for {}: failed, topic not owned .", topicPartition); + } registerPrepareMetadataFailedEvent(startPrepareMetadataNanos); // remove null future cache KafkaTopicConsumerManagerCache.getInstance().removeAndCloseByTopic(fullTopicName); @@ -491,7 +530,7 @@ private void handleEntries(final List entries, highWatermark, // TODO: should it be changed to the logStartOffset? abortedTransactions, kafkaRecords)); - bytesReadable.getAndAdd(kafkaRecords.sizeInBytes()); + bytesRead.getAndAdd(kafkaRecords.sizeInBytes()); tryComplete(); }); } diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/RequestStats.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/RequestStats.java index cff8d88a1c..99e79faa48 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/RequestStats.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/RequestStats.java @@ -28,6 +28,7 @@ import static io.streamnative.pulsar.handlers.kop.KopServerStats.RESPONSE_BLOCKED_LATENCY; import static io.streamnative.pulsar.handlers.kop.KopServerStats.RESPONSE_BLOCKED_TIMES; import static io.streamnative.pulsar.handlers.kop.KopServerStats.SERVER_SCOPE; +import static io.streamnative.pulsar.handlers.kop.KopServerStats.WAITING_FETCHES_TRIGGERED; import io.streamnative.pulsar.handlers.kop.stats.StatsLogger; import java.util.concurrent.atomic.AtomicInteger; @@ -112,6 +113,12 @@ public class RequestStats { ) private final OpStatsLogger fetchDecodeStats; + @StatsDoc( + name = WAITING_FETCHES_TRIGGERED, + help = "number of pending fetches that woke up due to some data produced" + ) + private final Counter waitingFetchesTriggered; + public RequestStats(StatsLogger statsLogger) { this.statsLogger = statsLogger; @@ -127,6 +134,7 @@ public RequestStats(StatsLogger statsLogger) { this.prepareMetadataStats = statsLogger.getOpStatsLogger(PREPARE_METADATA); this.messageReadStats = statsLogger.getOpStatsLogger(MESSAGE_READ); this.fetchDecodeStats = statsLogger.getOpStatsLogger(FETCH_DECODE); + this.waitingFetchesTriggered = statsLogger.getCounter(WAITING_FETCHES_TRIGGERED); statsLogger.registerGauge(REQUEST_QUEUE_SIZE, new Gauge() { @Override diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/delayed/DelayedOperation.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/delayed/DelayedOperation.java index b03234cc26..6ff448a26d 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/delayed/DelayedOperation.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/delayed/DelayedOperation.java @@ -98,6 +98,13 @@ public boolean isCompleted() { */ public abstract boolean tryComplete(); + /** + * Try to wake up the operation. + */ + public boolean wakeup() { + return true; + } + /** * Thread-safe variant of tryComplete() that attempts completion only if the lock can be acquired * without blocking. diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/delayed/DelayedOperationPurgatory.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/delayed/DelayedOperationPurgatory.java index fafa361754..e1d6ef5a0c 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/delayed/DelayedOperationPurgatory.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/delayed/DelayedOperationPurgatory.java @@ -226,7 +226,7 @@ public int checkAndComplete(Object key) { if (null == watchers) { return 0; } else { - return watchers.tryCompleteWatched(); + return watchers.tryCompleteWatched(key); } } @@ -340,7 +340,7 @@ public void watch(T t) { } // traverse the list and try to complete some watched elements - public int tryCompleteWatched() { + public int tryCompleteWatched(Object key) { int completed = 0; Iterator iter = operations.iterator(); @@ -349,7 +349,7 @@ public int tryCompleteWatched() { if (curr.isCompleted()) { // another thread has completed this operation, just remove it iter.remove(); - } else if (curr.maybeTryComplete()) { + } else if (curr.wakeup() && curr.maybeTryComplete()) { iter.remove(); completed += 1; } diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java index 530904a429..2887bb16f5 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java @@ -30,9 +30,11 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder.KafkaHeaderAndRequest; +import java.io.InputStream; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; import java.util.Collection; @@ -46,7 +48,12 @@ import java.util.stream.Collectors; import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.io.IOUtils; import org.apache.commons.lang3.tuple.Pair; +import org.apache.http.HttpResponse; +import org.apache.http.client.HttpClient; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.impl.client.HttpClientBuilder; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; @@ -380,17 +387,21 @@ public void testFetchMinBytes() throws Exception { int maxWaitMs = 3000; int minBytes = 1; // case1: consuming an empty topic. + @Cleanup KafkaConsumer consumer1 = createKafkaConsumer(maxWaitMs, minBytes); consumer1.assign(topicPartitions); Long startTime1 = System.currentTimeMillis(); - consumer1.poll(Duration.ofMillis(maxWaitMs)); + ConsumerRecords emptyResult = consumer1.poll(Duration.ofMillis(maxWaitMs)); Long endTime1 = System.currentTimeMillis(); log.info("cost time1:" + (endTime1 - startTime1)); + assertEquals(0, emptyResult.count()); // case2: consuming an topic after producing data. + @Cleanup KafkaProducer kProducer = createKafkaProducer(); produceData(kProducer, topicPartitions, 10); + @Cleanup KafkaConsumer consumer2 = createKafkaConsumer(maxWaitMs, minBytes); consumer2.assign(topicPartitions); consumer2.seekToBeginning(topicPartitions); @@ -407,6 +418,57 @@ public void testFetchMinBytes() throws Exception { assertTrue(endTime2 - startTime2 < maxWaitMs); } + /** + * Test the sending speed of fetch request when the readable data is less than fetch.minBytes. + */ + @Test(timeOut = 60000) + public void testFetchMinBytesSingleConsumer() throws Exception { + String topicName = "testMinBytesTopic"; + TopicPartition tp = new TopicPartition(topicName, 0); + + // create partitioned topic. + admin.topics().createPartitionedTopic(topicName, 1); + List topicPartitions = new ArrayList<>(); + topicPartitions.add(tp); + + int maxWaitMs = 3000; // very long time + int minBytes = 1; + // case1: consuming an empty topic. + @Cleanup + KafkaConsumer consumer1 = createKafkaConsumer(maxWaitMs, minBytes); + consumer1.assign(topicPartitions); + ConsumerRecords emptyResult = consumer1.poll(Duration.ofMillis(maxWaitMs)); + assertEquals(0, emptyResult.count()); + + // case2: consuming an topic after producing data. + @Cleanup + KafkaProducer kProducer = createKafkaProducer(); + produceData(kProducer, topicPartitions, 10); + + int totalRead = 0; + do { + // Consumer1 is able to eventually read the data + // please note that we are passing 100 as max pool time + ConsumerRecords goodResultFrom1 = consumer1.poll(Duration.ofMillis(100)); + totalRead += goodResultFrom1.count(); + log.info("read {} records totalRead {}", goodResultFrom1.count(), totalRead); + // we require that every pool returns at least one record + // that is that we NEVER hit the maxWait timeout and also the pool timeout + assertTrue(goodResultFrom1.count() > 0); + } while (totalRead < 10); + assertEquals(10, totalRead); + + + HttpClient httpClient = HttpClientBuilder.create().build(); + final String metricsEndPoint = pulsar.getWebServiceAddress() + "/metrics"; + HttpResponse response = httpClient.execute(new HttpGet(metricsEndPoint)); + InputStream inputStream = response.getEntity().getContent(); + String metrics = IOUtils.toString(inputStream, StandardCharsets.UTF_8); + log.info("metrics {}", metrics); + + assertTrue(metrics.contains("kop_server_WAITING_FETCHES_TRIGGERED 1")); + } + @Test(timeOut = 80000) public void testConsumerListOffset() throws Exception { String topicName = "listOffset";