You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/10/12 23:06:26 UTC
[GitHub] [kafka] hachikuji opened a new pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
hachikuji opened a new pull request #9418:
URL: https://github.com/apache/kafka/pull/9418
The main purpose of this patch is to add `quorum.linger.ms` behavior to the raft implementation. This gives users a powerful knob to tune the impact of fsync. When an append is accepted from the state machine, it is held in an accumulator (similar to the producer) until the configured linger time is exceeded. This allows the implementation to amortize fsync overhead at the expense of some write latency.
The patch also improves our methodology for testing performance. Up to now, we have relied on the producer performance test, but it is difficult to simulate expected controller loads because producer performance is limited by other factors such as the number of producer clients and head-of-line blocking. Instead, this patch adds a workload generator which runs on the leader after election.
Finally, this patch brings us nearer to the write semantics expected by the KIP-500 controller. It makes the following changes:
- Introduce `RecordSerde<T>` interface which abstracts the underlying log implementation from `RaftClient`. The generic type is carried over to `RaftClient<T>` and is exposed through the read/write APIs.
- `RaftClient.append` is changed to `RaftClient.scheduleAppend` and returns the last offset of the expected log append.
- `RaftClient.scheduleAppend` accepts a batch of list of records and ensures that the full set are included in a single batch.
- Introduce `RaftClient.Listener` with a single `handleCommit` API which will eventually replace `RaftClient.read` in order to surface committed data to the controller state machine. Currently `handleCommit` is only used for records appended by the leader.
(Note that this patch piggybacks on top of #9352. I will rebase as soon as this PR has been merged.)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509732236
##########
File path: core/src/main/scala/kafka/raft/KafkaNetworkChannel.scala
##########
@@ -216,11 +216,9 @@ class KafkaNetworkChannel(time: Time,
endpoints.put(id, node)
}
- def postInboundRequest(header: RequestHeader,
- request: AbstractRequest,
- onResponseReceived: ResponseHandler): Unit = {
+ def postInboundRequest(request: AbstractRequest, onResponseReceived: ResponseHandler): Unit = {
Review comment:
This was actually a bug. The implementation was treating `correlationId` as unique across all connections, which of course was wrong. My fix was just to overwrite the `correlationId` from the header with one that could be unique, but obviously this loses traceability through the Raft layer. If it's ok with you, I'd like to address this problem more generally in a follow-up.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509711302
##########
File path: core/src/main/scala/kafka/tools/TestRaftServer.scala
##########
@@ -272,7 +291,79 @@ class TestRaftServer(val config: KafkaConfig) extends Logging {
)
}
- class RaftIoThread(client: KafkaRaftClient) extends ShutdownableThread("raft-io-thread") {
+ class RaftWorkloadGenerator(
+ client: KafkaRaftClient[Array[Byte]],
+ time: Time,
+ brokerId: Int,
+ recordsPerSec: Int,
+ recordSize: Int
+ ) extends ShutdownableThread(name = "raft-workload-generator") with RaftClient.Listener[Array[Byte]] {
+
+ private val stats = new WriteStats(time, printIntervalMs = 5000)
+ private val payload = new Array[Byte](recordSize)
+ private val pendingAppends = new util.ArrayDeque[PendingAppend]()
+
+ private var latestLeaderAndEpoch = new LeaderAndEpoch(OptionalInt.empty, 0)
+ private var isLeader = false
+ private var throttler: ThroughputThrottler = _
+ private var recordCount = 0
+
+ override def doWork(): Unit = {
+ if (latestLeaderAndEpoch != client.currentLeaderAndEpoch()) {
+ latestLeaderAndEpoch = client.currentLeaderAndEpoch()
+ isLeader = latestLeaderAndEpoch.leaderId.orElse(-1) == brokerId
+ if (isLeader) {
+ pendingAppends.clear()
Review comment:
You are right that the appends may still be committed, but in this patch, the `handleCommit` API is only invoked for appends within the current epoch. I thought it seemed simpler for now to just reset state at the start of the epoch. We can be more clever in the future once `handleCommit` is extended to account for replication.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rajinisivaram commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
rajinisivaram commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r511842783
##########
File path: core/src/main/scala/kafka/common/RecordValidationException.scala
##########
@@ -23,5 +23,6 @@ import org.apache.kafka.common.requests.ProduceResponse.RecordError
import scala.collection.Seq
class RecordValidationException(val invalidException: ApiException,
- val recordErrors: Seq[RecordError]) extends RuntimeException {
+ val recordErrors: Seq[RecordError])
+ extends RuntimeException(invalidException) {
Review comment:
I had only changed client-side code, so changes here wouldn't have conflicted.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r506638105
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java
##########
@@ -0,0 +1,296 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.Closeable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * TODO: Also flush after minimum size limit is reached?
Review comment:
Yeah, I think we should do it. I will create a JIRA and leave this for a follow-up.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r508910810
##########
File path: clients/src/main/java/org/apache/kafka/common/protocol/DataOutputWritable.java
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.protocol;
+
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.common.utils.Utils;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class DataOutputWritable implements Writable {
Review comment:
nit: not clear why we want an intermediate class instead of just having a `DataOutputStreamWritable` class and its constructor parameter declared as `DataOutputStream`?
##########
File path: core/src/main/scala/kafka/common/RecordValidationException.scala
##########
@@ -23,5 +23,6 @@ import org.apache.kafka.common.requests.ProduceResponse.RecordError
import scala.collection.Seq
class RecordValidationException(val invalidException: ApiException,
- val recordErrors: Seq[RecordError]) extends RuntimeException {
+ val recordErrors: Seq[RecordError])
+ extends RuntimeException(invalidException) {
Review comment:
I vaguely remember @rajinisivaram has another PR which may be overlapping with this fix, will let her to check and confirm if it is fine.
##########
File path: core/src/main/scala/kafka/tools/TestRaftServer.scala
##########
@@ -19,37 +19,44 @@ package kafka.tools
import java.io.File
import java.nio.file.Files
-import java.util.concurrent.CountDownLatch
-import java.util.{Properties, Random}
+import java.util
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+import java.util.{Collections, OptionalInt, Random}
-import joptsimple.OptionParser
+import com.yammer.metrics.core.MetricName
+import joptsimple.OptionException
import kafka.log.{Log, LogConfig, LogManager}
import kafka.network.SocketServer
import kafka.raft.{KafkaFuturePurgatory, KafkaMetadataLog, KafkaNetworkChannel}
import kafka.security.CredentialProvider
import kafka.server.{BrokerTopicStats, KafkaConfig, KafkaRequestHandlerPool, KafkaServer, LogDirFailureChannel}
+import kafka.tools.TestRaftServer.{ByteArraySerde, PendingAppend, ThroughputThrottler, WriteStats}
Review comment:
Why we need to import the inner classes?
##########
File path: core/src/main/scala/kafka/tools/TestRaftServer.scala
##########
@@ -272,7 +291,79 @@ class TestRaftServer(val config: KafkaConfig) extends Logging {
)
}
- class RaftIoThread(client: KafkaRaftClient) extends ShutdownableThread("raft-io-thread") {
+ class RaftWorkloadGenerator(
+ client: KafkaRaftClient[Array[Byte]],
+ time: Time,
+ brokerId: Int,
+ recordsPerSec: Int,
+ recordSize: Int
+ ) extends ShutdownableThread(name = "raft-workload-generator") with RaftClient.Listener[Array[Byte]] {
+
+ private val stats = new WriteStats(time, printIntervalMs = 5000)
+ private val payload = new Array[Byte](recordSize)
+ private val pendingAppends = new util.ArrayDeque[PendingAppend]()
+
+ private var latestLeaderAndEpoch = new LeaderAndEpoch(OptionalInt.empty, 0)
+ private var isLeader = false
+ private var throttler: ThroughputThrottler = _
+ private var recordCount = 0
+
+ override def doWork(): Unit = {
+ if (latestLeaderAndEpoch != client.currentLeaderAndEpoch()) {
+ latestLeaderAndEpoch = client.currentLeaderAndEpoch()
+ isLeader = latestLeaderAndEpoch.leaderId.orElse(-1) == brokerId
+ if (isLeader) {
+ pendingAppends.clear()
+ throttler = new ThroughputThrottler(time, recordsPerSec)
+ recordCount = 0
+ }
+ }
+
+ if (isLeader) {
+ recordCount += 1
+
+ val startTimeMs = time.milliseconds()
+ val sendTimeMs = if (throttler.maybeThrottle(recordCount, startTimeMs)) {
+ time.milliseconds()
+ } else {
+ startTimeMs
+ }
+
+ val offset = client.scheduleAppend(latestLeaderAndEpoch.epoch, Collections.singletonList(payload))
+ if (offset == null || offset == Long.MaxValue) {
+ time.sleep(10)
+ } else {
+ pendingAppends.offer(PendingAppend(latestLeaderAndEpoch.epoch, offset, sendTimeMs))
+ }
+ } else {
+ time.sleep(500)
+ }
+ }
+
+ override def handleCommit(epoch: Int, lastOffset: Long, records: util.List[Array[Byte]]): Unit = {
+ var offset = lastOffset - records.size() + 1
+ val currentTimeMs = time.milliseconds()
+
+ for (record <- records.asScala) {
+ val pendingAppend = pendingAppends.poll()
+ if (pendingAppend.epoch != epoch || pendingAppend.offset!= offset) {
+ warn(s"Expected next commit at offset ${pendingAppend.offset}, " +
Review comment:
Should this ever happen? If not, then we could shutdown the server after logging the error.
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1605,100 +1708,18 @@ public void poll() throws IOException {
}
}
- private void failPendingAppends(KafkaException exception) {
- for (UnwrittenAppend unwrittenAppend : unwrittenAppends) {
- unwrittenAppend.fail(exception);
- }
- unwrittenAppends.clear();
- }
-
- private void pollPendingAppends(LeaderState state, long currentTimeMs) {
- int numAppends = 0;
- int maxNumAppends = unwrittenAppends.size();
-
- while (!unwrittenAppends.isEmpty() && numAppends < maxNumAppends) {
- final UnwrittenAppend unwrittenAppend = unwrittenAppends.poll();
-
- if (unwrittenAppend.future.isDone())
- continue;
-
- if (unwrittenAppend.isTimedOut(currentTimeMs)) {
- unwrittenAppend.fail(new TimeoutException("Request timeout " + unwrittenAppend.requestTimeoutMs
- + " expired before the records could be appended to the log"));
- } else {
- int epoch = quorum.epoch();
- LogAppendInfo info = appendAsLeader(unwrittenAppend.records);
- OffsetAndEpoch offsetAndEpoch = new OffsetAndEpoch(info.lastOffset, epoch);
- long numRecords = info.lastOffset - info.firstOffset + 1;
- logger.debug("Completed write of {} records at {}", numRecords, offsetAndEpoch);
-
- if (unwrittenAppend.ackMode == AckMode.LEADER) {
- unwrittenAppend.complete(offsetAndEpoch);
- } else if (unwrittenAppend.ackMode == AckMode.QUORUM) {
- CompletableFuture<Long> future = appendPurgatory.await(
- LogOffset.awaitCommitted(offsetAndEpoch.offset),
- unwrittenAppend.requestTimeoutMs);
-
- future.whenComplete((completionTimeMs, exception) -> {
- if (exception != null) {
- logger.error("Failed to commit append at {} due to {}", offsetAndEpoch, exception);
-
- unwrittenAppend.fail(exception);
- } else {
- long elapsedTime = Math.max(0, completionTimeMs - currentTimeMs);
- double elapsedTimePerRecord = (double) elapsedTime / numRecords;
- kafkaRaftMetrics.updateCommitLatency(elapsedTimePerRecord, currentTimeMs);
- unwrittenAppend.complete(offsetAndEpoch);
-
- logger.debug("Completed commit of {} records at {}", numRecords, offsetAndEpoch);
- }
- });
- }
- }
-
- numAppends++;
- }
-
- if (numAppends > 0) {
- flushLeaderLog(state, currentTimeMs);
- }
- }
-
- /**
- * Append a set of records to the log. Successful completion of the future indicates a success of
- * the append, with the uncommitted base offset and epoch.
- *
- * @param records The records to write to the log
- * @param ackMode The commit mode for the appended records
- * @param timeoutMs The maximum time to wait for the append operation to complete (including
- * any time needed for replication)
- * @return The uncommitted base offset and epoch of the appended records
- */
@Override
- public CompletableFuture<OffsetAndEpoch> append(
- Records records,
- AckMode ackMode,
- long timeoutMs
- ) {
- if (records.sizeInBytes() == 0)
- throw new IllegalArgumentException("Attempt to append empty record set");
-
- if (shutdown.get() != null)
- throw new IllegalStateException("Cannot append records while we are shutting down");
-
- if (quorum.isObserver())
- throw new IllegalStateException("Illegal attempt to write to an observer");
-
- CompletableFuture<OffsetAndEpoch> future = new CompletableFuture<>();
- UnwrittenAppend unwrittenAppend = new UnwrittenAppend(
- records, time.milliseconds(), timeoutMs, ackMode, future);
+ public Long scheduleAppend(int epoch, List<T> records) {
+ BatchAccumulator<T> accumulator = this.accumulator;
+ if (accumulator == null) {
+ return Long.MAX_VALUE;
Review comment:
Why return MAX_VALUE instead of null here? If we want to use `null` to indicate `memory full` and use `MAX_VALUE` to indicate `not leader`, the javadoc should reflecting this.
Anyways, I think returning sth like a `combo(Offset, ErrorCode, backoffMs)` would be preferred in the end state.
##########
File path: core/src/main/scala/kafka/tools/TestRaftRequestHandler.scala
##########
@@ -56,73 +47,8 @@ class TestRaftRequestHandler(
| ApiKeys.END_QUORUM_EPOCH
| ApiKeys.FETCH =>
val requestBody = request.body[AbstractRequest]
- networkChannel.postInboundRequest(
- request.header,
- requestBody,
- response => sendResponse(request, Some(response)))
-
- case ApiKeys.API_VERSIONS =>
Review comment:
Could you elaborate a bit why we would not expect these api keys in the test handler any more?
##########
File path: core/src/main/scala/kafka/raft/KafkaNetworkChannel.scala
##########
@@ -216,11 +216,9 @@ class KafkaNetworkChannel(time: Time,
endpoints.put(id, node)
}
- def postInboundRequest(header: RequestHeader,
- request: AbstractRequest,
- onResponseReceived: ResponseHandler): Unit = {
+ def postInboundRequest(request: AbstractRequest, onResponseReceived: ResponseHandler): Unit = {
Review comment:
nit: maybe add a comment indicating this function is for testing only (and hence we do not care about the correlation id).
##########
File path: core/src/main/scala/kafka/tools/TestRaftServer.scala
##########
@@ -272,7 +291,79 @@ class TestRaftServer(val config: KafkaConfig) extends Logging {
)
}
- class RaftIoThread(client: KafkaRaftClient) extends ShutdownableThread("raft-io-thread") {
+ class RaftWorkloadGenerator(
+ client: KafkaRaftClient[Array[Byte]],
+ time: Time,
+ brokerId: Int,
+ recordsPerSec: Int,
+ recordSize: Int
+ ) extends ShutdownableThread(name = "raft-workload-generator") with RaftClient.Listener[Array[Byte]] {
+
+ private val stats = new WriteStats(time, printIntervalMs = 5000)
+ private val payload = new Array[Byte](recordSize)
+ private val pendingAppends = new util.ArrayDeque[PendingAppend]()
+
+ private var latestLeaderAndEpoch = new LeaderAndEpoch(OptionalInt.empty, 0)
+ private var isLeader = false
+ private var throttler: ThroughputThrottler = _
+ private var recordCount = 0
+
+ override def doWork(): Unit = {
+ if (latestLeaderAndEpoch != client.currentLeaderAndEpoch()) {
+ latestLeaderAndEpoch = client.currentLeaderAndEpoch()
+ isLeader = latestLeaderAndEpoch.leaderId.orElse(-1) == brokerId
+ if (isLeader) {
+ pendingAppends.clear()
+ throttler = new ThroughputThrottler(time, recordsPerSec)
+ recordCount = 0
+ }
+ }
+
+ if (isLeader) {
+ recordCount += 1
+
+ val startTimeMs = time.milliseconds()
+ val sendTimeMs = if (throttler.maybeThrottle(recordCount, startTimeMs)) {
+ time.milliseconds()
+ } else {
+ startTimeMs
+ }
+
+ val offset = client.scheduleAppend(latestLeaderAndEpoch.epoch, Collections.singletonList(payload))
+ if (offset == null || offset == Long.MaxValue) {
Review comment:
If `offset == Long.MaxValue` would sleeping save us anything? Should we terminate the testing server then?
##########
File path: core/src/main/scala/kafka/tools/TestRaftServer.scala
##########
@@ -272,7 +291,79 @@ class TestRaftServer(val config: KafkaConfig) extends Logging {
)
}
- class RaftIoThread(client: KafkaRaftClient) extends ShutdownableThread("raft-io-thread") {
+ class RaftWorkloadGenerator(
+ client: KafkaRaftClient[Array[Byte]],
+ time: Time,
+ brokerId: Int,
+ recordsPerSec: Int,
+ recordSize: Int
+ ) extends ShutdownableThread(name = "raft-workload-generator") with RaftClient.Listener[Array[Byte]] {
+
+ private val stats = new WriteStats(time, printIntervalMs = 5000)
+ private val payload = new Array[Byte](recordSize)
+ private val pendingAppends = new util.ArrayDeque[PendingAppend]()
+
+ private var latestLeaderAndEpoch = new LeaderAndEpoch(OptionalInt.empty, 0)
+ private var isLeader = false
+ private var throttler: ThroughputThrottler = _
+ private var recordCount = 0
+
+ override def doWork(): Unit = {
+ if (latestLeaderAndEpoch != client.currentLeaderAndEpoch()) {
+ latestLeaderAndEpoch = client.currentLeaderAndEpoch()
+ isLeader = latestLeaderAndEpoch.leaderId.orElse(-1) == brokerId
+ if (isLeader) {
+ pendingAppends.clear()
Review comment:
Even if the leader has changed back, previously pending appends may still get committed right? I think it is sufficient to just poll-and-drop them in `handleCommit` when leader changes and not clearing the queue here.
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -148,36 +162,44 @@ public KafkaRaftClient(RaftConfig raftConfig,
raftConfig.retryBackoffMs(),
raftConfig.requestTimeoutMs(),
1000,
+ raftConfig.lingerMs(),
logContext,
new Random());
}
- public KafkaRaftClient(NetworkChannel channel,
- ReplicatedLog log,
- QuorumState quorum,
- Time time,
- Metrics metrics,
- FuturePurgatory<LogOffset> fetchPurgatory,
- FuturePurgatory<LogOffset> appendPurgatory,
- Map<Integer, InetSocketAddress> voterAddresses,
- int electionBackoffMaxMs,
- int retryBackoffMs,
- int requestTimeoutMs,
- int fetchMaxWaitMs,
- LogContext logContext,
- Random random) {
+ public KafkaRaftClient(
+ RecordSerde<T> serde,
+ NetworkChannel channel,
+ ReplicatedLog log,
+ QuorumState quorum,
+ MemoryPool memoryPool,
+ Time time,
+ Metrics metrics,
+ FuturePurgatory<LogOffset> fetchPurgatory,
+ FuturePurgatory<LogOffset> appendPurgatory,
+ Map<Integer, InetSocketAddress> voterAddresses,
+ int electionBackoffMaxMs,
+ int retryBackoffMs,
+ int requestTimeoutMs,
+ int fetchMaxWaitMs,
+ int lingerMs,
Review comment:
nit: `appendLingerMs`?
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1443,15 +1485,79 @@ private void pollShutdown(GracefulShutdown shutdown) throws IOException {
}
}
+ private void appendBatch(
+ LeaderState state,
+ BatchAccumulator.CompletedBatch<T> batch,
+ long appendTimeMs
+ ) {
+ try {
+ List<T> records = batch.records;
+ int epoch = state.epoch();
+
+ LogAppendInfo info = appendAsLeader(batch.data);
+ OffsetAndEpoch offsetAndEpoch = new OffsetAndEpoch(info.lastOffset, epoch);
+ CompletableFuture<Long> future = appendPurgatory.await(
+ LogOffset.awaitCommitted(offsetAndEpoch.offset),
+ Integer.MAX_VALUE
+ );
+
+ future.whenComplete((commitTimeMs, exception) -> {
+ int numRecords = batch.records.size();
+ if (exception != null) {
+ logger.debug("Failed to commit {} records at {}", numRecords, offsetAndEpoch, exception);
+ } else {
+ long elapsedTime = Math.max(0, commitTimeMs - appendTimeMs);
+ double elapsedTimePerRecord = (double) elapsedTime / numRecords;
+ kafkaRaftMetrics.updateCommitLatency(elapsedTimePerRecord, appendTimeMs);
+ logger.debug("Completed commit of {} records at {}", numRecords, offsetAndEpoch);
+ listener.handleCommit(epoch, info.lastOffset, records);
+ }
+ });
+ } finally {
+ batch.release();
+ }
+ }
+
+ private long maybeAppendBatches(
+ LeaderState state,
+ long currentTimeMs
+ ) {
+ long timeUnitFlush = accumulator.timeUntilFlush(currentTimeMs);
+ if (timeUnitFlush <= 0) {
+ List<BatchAccumulator.CompletedBatch<T>> batches = accumulator.flush();
+ Iterator<BatchAccumulator.CompletedBatch<T>> iterator = batches.iterator();
+
+ try {
+ while (iterator.hasNext()) {
+ BatchAccumulator.CompletedBatch<T> batch = iterator.next();
+ appendBatch(state, batch, currentTimeMs);
+ }
+ flushLeaderLog(state, currentTimeMs);
Review comment:
For now since we only have a single thread processing all incoming req/resp, this is okay; but when we multi-thread processing requests this would no longer be safe, since it is possible that some batches gets replicated and committed while not being flushed locally yet.
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchBuilder.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.protocol.DataOutputStreamWritable;
+import org.apache.kafka.common.record.AbstractRecords;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.DefaultRecord;
+import org.apache.kafka.common.record.DefaultRecordBatch;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.DataOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class BatchBuilder<T> {
Review comment:
nit: can we consolidate this class with the existing one (of course, it needs to be generalized on T record) than creating a new class? Ditto for BatchAccumulator and BatchMemoryPool.
##########
File path: raft/src/main/java/org/apache/kafka/raft/ReplicatedCounter.java
##########
@@ -104,29 +62,17 @@ public synchronized boolean isWritable() {
public synchronized void increment() {
if (!isWritable())
throw new KafkaException("Counter is not currently writable");
- int initialValue = uncommitted.get();
- int incrementedValue = uncommitted.incrementAndGet();
- Records records = MemoryRecords.withRecords(CompressionType.NONE, serialize(incrementedValue));
- client.append(records, AckMode.LEADER, Integer.MAX_VALUE).whenComplete((offsetAndEpoch, throwable) -> {
- if (offsetAndEpoch != null) {
- log.debug("Appended increment at offset {}: {} -> {}",
- offsetAndEpoch.offset, initialValue, incrementedValue);
- } else {
- uncommitted.set(initialValue);
- log.debug("Failed append of increment: {} -> {}", initialValue, incrementedValue, throwable);
- }
- });
- }
-
- private SimpleRecord serialize(int value) {
- ByteBuffer buffer = ByteBuffer.allocate(4);
- Type.INT32.write(buffer, value);
- buffer.flip();
- return new SimpleRecord(buffer);
+ uncommitted += 1;
+ long offset = client.scheduleAppend(currentLeaderAndEpoch.epoch, Collections.singletonList(uncommitted));
Review comment:
Should we handle `null / max-value` here?
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java
##########
@@ -0,0 +1,296 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.Closeable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * TODO: Also flush after minimum size limit is reached?
+ */
+public class BatchAccumulator<T> implements Closeable {
+ private final int epoch;
+ private final Time time;
+ private final Timer lingerTimer;
+ private final int lingerMs;
+ private final int maxBatchSize;
+ private final CompressionType compressionType;
+ private final MemoryPool memoryPool;
+ private final ReentrantLock lock;
+ private final RecordSerde<T> serde;
+
+ private long nextOffset;
+ private BatchBuilder<T> currentBatch;
+ private List<CompletedBatch<T>> completed;
+
+ public BatchAccumulator(
+ int epoch,
+ long baseOffset,
+ int lingerMs,
+ int maxBatchSize,
+ MemoryPool memoryPool,
+ Time time,
+ CompressionType compressionType,
+ RecordSerde<T> serde
+ ) {
+ this.epoch = epoch;
+ this.lingerMs = lingerMs;
+ this.maxBatchSize = maxBatchSize;
+ this.memoryPool = memoryPool;
+ this.time = time;
+ this.lingerTimer = time.timer(lingerMs);
+ this.compressionType = compressionType;
+ this.serde = serde;
+ this.nextOffset = baseOffset;
+ this.completed = new ArrayList<>();
+ this.lock = new ReentrantLock();
+ }
+
+ /**
+ * Append a list of records into an atomic batch. We guarantee all records
+ * are included in the same underlying record batch so that either all of
+ * the records become committed or none of them do.
+ *
+ * @param epoch the expected leader epoch
+ * @param records the list of records to include in a batch
+ * @return the offset of the last message or {@link Long#MAX_VALUE} if the epoch
+ * does not match
+ */
+ public Long append(int epoch, List<T> records) {
+ if (epoch != this.epoch) {
+ // If the epoch does not match, then the state machine probably
+ // has not gotten the notification about the latest epoch change.
+ // In this case, ignore the append and return a large offset value
+ // which will never be committed.
+ return Long.MAX_VALUE;
+ }
+
+ Object serdeContext = serde.newWriteContext();
+ int batchSize = 0;
+ for (T record : records) {
+ batchSize += serde.recordSize(record, serdeContext);
+ }
+
+ if (batchSize > maxBatchSize) {
+ throw new IllegalArgumentException("The total size of " + records + " is " + batchSize +
+ ", which exceeds the maximum allowed batch size of " + maxBatchSize);
+ }
+
+ lock.lock();
+ try {
+ BatchBuilder<T> batch = maybeAllocateBatch(batchSize);
+ if (batch == null) {
+ return null;
+ }
+
+ if (isEmpty()) {
+ lingerTimer.update();
+ lingerTimer.reset(lingerMs);
+ }
+
+ for (T record : records) {
+ batch.appendRecord(record, serdeContext);
+ nextOffset += 1;
+ }
+
+ return nextOffset - 1;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private BatchBuilder<T> maybeAllocateBatch(int batchSize) {
+ if (currentBatch == null) {
+ startNewBatch();
+ } else if (!currentBatch.hasRoomFor(batchSize)) {
+ completeCurrentBatch();
+ }
+ return currentBatch;
+ }
+
+ private void completeCurrentBatch() {
+ MemoryRecords data = currentBatch.build();
+ completed.add(new CompletedBatch<>(
+ currentBatch.baseOffset(),
+ currentBatch.records(),
+ data,
+ memoryPool,
+ currentBatch.initialBuffer()
+ ));
+ currentBatch = null;
+ startNewBatch();
+ }
+
+ private void startNewBatch() {
+ ByteBuffer buffer = memoryPool.tryAllocate(maxBatchSize);
+ if (buffer != null) {
+ currentBatch = new BatchBuilder<>(
+ buffer,
+ serde,
+ compressionType,
+ nextOffset,
+ time.milliseconds(),
+ false,
+ RecordBatch.NO_PARTITION_LEADER_EPOCH,
+ maxBatchSize
+ );
+ }
+ }
+
+ /**
+ * Check whether there are any batches which need flushing now.
+ *
+ * @param currentTimeMs current time in milliseconds
+ * @return true if there are batches ready to flush, false otherwise
+ */
+ public boolean needsFlush(long currentTimeMs) {
+ return timeUntilFlush(currentTimeMs) <= 0;
+ }
+
+ /**
+ * Check the time remaining until the next needed flush. If the accumulator
+ * is empty, then {@link Long#MAX_VALUE} will be returned.
+ *
+ * @param currentTimeMs current time in milliseconds
+ * @return the delay in milliseconds before the next expected flush
+ */
+ public long timeUntilFlush(long currentTimeMs) {
+ lock.lock();
+ try {
+ lingerTimer.update(currentTimeMs);
+ if (isEmpty()) {
+ return Long.MAX_VALUE;
+ } else {
+ return lingerTimer.remainingMs();
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private boolean isEmpty() {
+ lock.lock();
+ try {
+ if (currentBatch != null && currentBatch.nonEmpty()) {
+ return false;
+ } else {
+ return completed.isEmpty();
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Get the leader epoch, which is constant for each instance.
+ *
+ * @return the leader epoch
+ */
+ public int epoch() {
+ return epoch;
+ }
+
+ /**
+ * Flush completed batches. The caller is expected to first check whether
+ * a flush is expected using {@link #needsFlush(long)} in order to avoid
+ * unnecessary flushing.
+ *
+ * @return the list of completed batches
+ */
+ public List<CompletedBatch<T>> flush() {
+ lock.lock();
+ try {
+ if (currentBatch != null && currentBatch.nonEmpty()) {
+ completeCurrentBatch();
+ }
+
+ List<CompletedBatch<T>> res = completed;
+ this.completed = new ArrayList<>();
+ return res;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Get the number of batches including the one that is currently being
+ * written to (if it exists).
+ *
+ * @return
+ */
+ public int count() {
Review comment:
Is this going to be used for non-testing code in the future? If it is only going to be for metrics purposes maybe we can allow it to be non thread-safe just to not blocking on other critical paths.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509615681
##########
File path: core/src/main/scala/kafka/tools/TestRaftRequestHandler.scala
##########
@@ -56,73 +47,8 @@ class TestRaftRequestHandler(
| ApiKeys.END_QUORUM_EPOCH
| ApiKeys.FETCH =>
val requestBody = request.body[AbstractRequest]
- networkChannel.postInboundRequest(
- request.header,
- requestBody,
- response => sendResponse(request, Some(response)))
-
- case ApiKeys.API_VERSIONS =>
Review comment:
I felt it was too difficult to approximate a controller workload using one or more producers because of the fact that we can only handle one request at a time. So I created a separate workload generator which executes on the leader and I removed all of this somewhat hacky handling logic which allowed us to use a producer.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #9418:
URL: https://github.com/apache/kafka/pull/9418
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509740150
##########
File path: core/src/main/scala/kafka/common/RecordValidationException.scala
##########
@@ -23,5 +23,6 @@ import org.apache.kafka.common.requests.ProduceResponse.RecordError
import scala.collection.Seq
class RecordValidationException(val invalidException: ApiException,
- val recordErrors: Seq[RecordError]) extends RuntimeException {
+ val recordErrors: Seq[RecordError])
+ extends RuntimeException(invalidException) {
Review comment:
Yeah, doesn't need to be done here. I just noticed the trace was missing when debugging a failure.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r512189731
##########
File path: core/src/main/scala/kafka/raft/KafkaNetworkChannel.scala
##########
@@ -216,11 +216,9 @@ class KafkaNetworkChannel(time: Time,
endpoints.put(id, node)
}
- def postInboundRequest(header: RequestHeader,
- request: AbstractRequest,
- onResponseReceived: ResponseHandler): Unit = {
+ def postInboundRequest(request: AbstractRequest, onResponseReceived: ResponseHandler): Unit = {
Review comment:
> Unrelated with this change but when reading the code. I was confused by the name pollInboundResponses. That function returns both pending request and responses.
I don't think that's right. It only delivers responses for outbound requests (in other words, inbound responses).
In regard to the confusion, I took an unconventional turn in the beginning by consolidating inbound and outbound requests through a single API. This was useful because it allowed inputs and outputs to be serialized, which made it easier to have deterministic simulation tests. The main defect from the perspective of Raft is that fetch requests cannot be handled concurrently. I think we are going to have to address this problem in the long term, but it is outside the scope of this PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r512217218
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java
##########
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.Closeable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * TODO: Also flush after minimum size limit is reached?
+ */
+public class BatchAccumulator<T> implements Closeable {
+ private final int epoch;
+ private final Time time;
+ private final Timer lingerTimer;
+ private final int lingerMs;
+ private final int maxBatchSize;
+ private final CompressionType compressionType;
+ private final MemoryPool memoryPool;
+ private final ReentrantLock lock;
+ private final RecordSerde<T> serde;
+
+ private long nextOffset;
+ private BatchBuilder<T> currentBatch;
+ private List<CompletedBatch<T>> completed;
+
+ public BatchAccumulator(
+ int epoch,
+ long baseOffset,
+ int lingerMs,
+ int maxBatchSize,
+ MemoryPool memoryPool,
+ Time time,
+ CompressionType compressionType,
+ RecordSerde<T> serde
+ ) {
+ this.epoch = epoch;
+ this.lingerMs = lingerMs;
+ this.maxBatchSize = maxBatchSize;
+ this.memoryPool = memoryPool;
+ this.time = time;
+ this.lingerTimer = time.timer(lingerMs);
+ this.compressionType = compressionType;
+ this.serde = serde;
+ this.nextOffset = baseOffset;
+ this.completed = new ArrayList<>();
+ this.lock = new ReentrantLock();
+ }
+
+ /**
+ * Append a list of records into an atomic batch. We guarantee all records
+ * are included in the same underlying record batch so that either all of
+ * the records become committed or none of them do.
+ *
+ * @param epoch the expected leader epoch
+ * @param records the list of records to include in a batch
+ * @return the offset of the last message or {@link Long#MAX_VALUE} if the epoch
+ * does not match
+ */
+ public Long append(int epoch, List<T> records) {
+ if (epoch != this.epoch) {
+ // If the epoch does not match, then the state machine probably
+ // has not gotten the notification about the latest epoch change.
+ // In this case, ignore the append and return a large offset value
+ // which will never be committed.
+ return Long.MAX_VALUE;
+ }
+
+ Object serdeContext = serde.newWriteContext();
+ int batchSize = 0;
+ for (T record : records) {
+ batchSize += serde.recordSize(record, serdeContext);
+ }
+
+ if (batchSize > maxBatchSize) {
+ throw new IllegalArgumentException("The total size of " + records + " is " + batchSize +
+ ", which exceeds the maximum allowed batch size of " + maxBatchSize);
+ }
+
+ lock.lock();
+ try {
+ BatchBuilder<T> batch = maybeAllocateBatch(batchSize);
+ if (batch == null) {
+ return null;
+ }
+
+ if (isEmpty()) {
+ lingerTimer.update();
+ lingerTimer.reset(lingerMs);
+ }
+
+ for (T record : records) {
+ batch.appendRecord(record, serdeContext);
+ nextOffset += 1;
+ }
+
+ return nextOffset - 1;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private BatchBuilder<T> maybeAllocateBatch(int batchSize) {
+ if (currentBatch == null) {
+ startNewBatch();
+ } else if (!currentBatch.hasRoomFor(batchSize)) {
+ completeCurrentBatch();
+ }
+ return currentBatch;
+ }
+
+ private void completeCurrentBatch() {
+ MemoryRecords data = currentBatch.build();
+ completed.add(new CompletedBatch<>(
+ currentBatch.baseOffset(),
+ currentBatch.records(),
+ data,
+ memoryPool,
+ currentBatch.initialBuffer()
+ ));
+ currentBatch = null;
+ startNewBatch();
+ }
+
+ private void startNewBatch() {
+ ByteBuffer buffer = memoryPool.tryAllocate(maxBatchSize);
+ if (buffer != null) {
+ currentBatch = new BatchBuilder<>(
+ buffer,
+ serde,
+ compressionType,
+ nextOffset,
+ time.milliseconds(),
+ false,
+ RecordBatch.NO_PARTITION_LEADER_EPOCH,
+ maxBatchSize
+ );
+ }
+ }
+
+ /**
+ * Check whether there are any batches which need flushing now.
Review comment:
Perhaps `drain` is a better verb in this context.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#issuecomment-717420938
LGTM.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509714949
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1443,15 +1485,79 @@ private void pollShutdown(GracefulShutdown shutdown) throws IOException {
}
}
+ private void appendBatch(
+ LeaderState state,
+ BatchAccumulator.CompletedBatch<T> batch,
+ long appendTimeMs
+ ) {
+ try {
+ List<T> records = batch.records;
+ int epoch = state.epoch();
+
+ LogAppendInfo info = appendAsLeader(batch.data);
+ OffsetAndEpoch offsetAndEpoch = new OffsetAndEpoch(info.lastOffset, epoch);
+ CompletableFuture<Long> future = appendPurgatory.await(
+ LogOffset.awaitCommitted(offsetAndEpoch.offset),
+ Integer.MAX_VALUE
+ );
+
+ future.whenComplete((commitTimeMs, exception) -> {
+ int numRecords = batch.records.size();
+ if (exception != null) {
+ logger.debug("Failed to commit {} records at {}", numRecords, offsetAndEpoch, exception);
+ } else {
+ long elapsedTime = Math.max(0, commitTimeMs - appendTimeMs);
+ double elapsedTimePerRecord = (double) elapsedTime / numRecords;
+ kafkaRaftMetrics.updateCommitLatency(elapsedTimePerRecord, appendTimeMs);
+ logger.debug("Completed commit of {} records at {}", numRecords, offsetAndEpoch);
+ listener.handleCommit(epoch, info.lastOffset, records);
+ }
+ });
+ } finally {
+ batch.release();
+ }
+ }
+
+ private long maybeAppendBatches(
+ LeaderState state,
+ long currentTimeMs
+ ) {
+ long timeUnitFlush = accumulator.timeUntilFlush(currentTimeMs);
+ if (timeUnitFlush <= 0) {
+ List<BatchAccumulator.CompletedBatch<T>> batches = accumulator.flush();
+ Iterator<BatchAccumulator.CompletedBatch<T>> iterator = batches.iterator();
+
+ try {
+ while (iterator.hasNext()) {
+ BatchAccumulator.CompletedBatch<T> batch = iterator.next();
+ appendBatch(state, batch, currentTimeMs);
+ }
+ flushLeaderLog(state, currentTimeMs);
Review comment:
Yes, I agree with you. Of course it is ok if unflushed data gets replicated. The main thing we need to protect is incrementing the high watermark.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r504344128
##########
File path: raft/src/main/java/org/apache/kafka/raft/RaftClient.java
##########
@@ -19,33 +19,51 @@
import org.apache.kafka.common.record.Records;
import java.io.IOException;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
-public interface RaftClient {
+public interface RaftClient<T> {
+
+ interface Listener<T> {
+ /**
+ * Callback which is invoked when records written through {@link #scheduleAppend(int, List)}
+ * become committed.
+ *
+ * Note that there is not a one-to-one correspondence between writes through
+ * {@link #scheduleAppend(int, List)} and this callback. The Raft implementation
+ * is free to batch together the records from multiple append calls provided
+ * that batch boundaries are respected. This means that each batch specified
+ * through {@link #scheduleAppend(int, List)} is guaranteed to be a subset of
+ * a batch passed to {@link #handleCommit(int, long, List)}.
+ *
+ * @param epoch the epoch in which the write was accepted
+ * @param lastOffset the last offset of the recor
Review comment:
fix typo
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r506636227
##########
File path: raft/src/main/java/org/apache/kafka/raft/RaftClient.java
##########
@@ -19,33 +19,51 @@
import org.apache.kafka.common.record.Records;
import java.io.IOException;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
-public interface RaftClient {
+public interface RaftClient<T> {
+
+ interface Listener<T> {
+ /**
+ * Callback which is invoked when records written through {@link #scheduleAppend(int, List)}
+ * become committed.
+ *
+ * Note that there is not a one-to-one correspondence between writes through
+ * {@link #scheduleAppend(int, List)} and this callback. The Raft implementation
+ * is free to batch together the records from multiple append calls provided
+ * that batch boundaries are respected. This means that each batch specified
+ * through {@link #scheduleAppend(int, List)} is guaranteed to be a subset of
+ * a batch passed to {@link #handleCommit(int, long, List)}.
+ *
+ * @param epoch the epoch in which the write was accepted
+ * @param lastOffset the offset of the last record in the record list
+ * @param records the set of records that were committed
+ */
+ void handleCommit(int epoch, long lastOffset, List<T> records);
+ }
/**
* Initialize the client. This should only be called once and it must be
* called before any of the other APIs can be invoked.
*
* @throws IOException For any IO errors during initialization
*/
- void initialize() throws IOException;
+ void initialize(Listener<T> listener) throws IOException;
/**
- * Append a new entry to the log. The client must be in the leader state to
- * accept an append: it is up to the state machine implementation
- * to ensure this using {@link #currentLeaderAndEpoch()}.
- *
- * TODO: One improvement we can make here is to allow the caller to specify
- * the current leader epoch in the record set. That would ensure that each
- * leader change must be "observed" by the state machine before new appends
- * are accepted.
+ * Append a list of records to the log. The write will be scheduled for some time
+ * in the future. There is no guarantee that appended records will be written to
+ * the log and eventually committed. However, it is guaranteed that if any of the
+ * records become committed, then all of them will be.
*
- * @param records The records to append to the log
- * @param timeoutMs Maximum time to wait for the append to complete
- * @return A future containing the last offset and epoch of the appended records (if successful)
+ * @param epoch the current leader epoch
+ * @param records the list of records to append
+ * @return the offset within the current epoch that the log entries will be appended,
+ * or null if the leader was unable to accept the write (e.g. due to memory
+ * being reached).
*/
- CompletableFuture<OffsetAndEpoch> append(Records records, AckMode ackMode, long timeoutMs);
+ Long scheduleAppend(int epoch, List<T> records);
Review comment:
Yeah, this was more of a workaround until we thought of something better. The current idea we are thinking about is letting the Raft layer return a backoff time, so the type would semantically be `Either[Offset, BackoffMs]`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r512847195
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchBuilder.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.protocol.DataOutputStreamWritable;
+import org.apache.kafka.common.record.AbstractRecords;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.DefaultRecord;
+import org.apache.kafka.common.record.DefaultRecordBatch;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.DataOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class BatchBuilder<T> {
Review comment:
Yes, that is fair. Initially I had planned to reuse as much as possible, but the existing classes had too much baggage. I wanted implementations which were simple and efficient for the needs of the raft layer.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r505036063
##########
File path: raft/src/main/java/org/apache/kafka/raft/RaftClient.java
##########
@@ -19,33 +19,51 @@
import org.apache.kafka.common.record.Records;
import java.io.IOException;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
-public interface RaftClient {
+public interface RaftClient<T> {
+
+ interface Listener<T> {
+ /**
+ * Callback which is invoked when records written through {@link #scheduleAppend(int, List)}
+ * become committed.
+ *
+ * Note that there is not a one-to-one correspondence between writes through
+ * {@link #scheduleAppend(int, List)} and this callback. The Raft implementation
+ * is free to batch together the records from multiple append calls provided
+ * that batch boundaries are respected. This means that each batch specified
+ * through {@link #scheduleAppend(int, List)} is guaranteed to be a subset of
+ * a batch passed to {@link #handleCommit(int, long, List)}.
+ *
+ * @param epoch the epoch in which the write was accepted
+ * @param lastOffset the offset of the last record in the record list
+ * @param records the set of records that were committed
+ */
+ void handleCommit(int epoch, long lastOffset, List<T> records);
+ }
/**
* Initialize the client. This should only be called once and it must be
* called before any of the other APIs can be invoked.
*
* @throws IOException For any IO errors during initialization
*/
- void initialize() throws IOException;
+ void initialize(Listener<T> listener) throws IOException;
/**
- * Append a new entry to the log. The client must be in the leader state to
- * accept an append: it is up to the state machine implementation
- * to ensure this using {@link #currentLeaderAndEpoch()}.
- *
- * TODO: One improvement we can make here is to allow the caller to specify
- * the current leader epoch in the record set. That would ensure that each
- * leader change must be "observed" by the state machine before new appends
- * are accepted.
+ * Append a list of records to the log. The write will be scheduled for some time
+ * in the future. There is no guarantee that appended records will be written to
+ * the log and eventually committed. However, it is guaranteed that if any of the
+ * records become committed, then all of them will be.
*
- * @param records The records to append to the log
- * @param timeoutMs Maximum time to wait for the append to complete
- * @return A future containing the last offset and epoch of the appended records (if successful)
+ * @param epoch the current leader epoch
+ * @param records the list of records to append
+ * @return the offset within the current epoch that the log entries will be appended,
+ * or null if the leader was unable to accept the write (e.g. due to memory
+ * being reached).
*/
- CompletableFuture<OffsetAndEpoch> append(Records records, AckMode ackMode, long timeoutMs);
+ Long scheduleAppend(int epoch, List<T> records);
Review comment:
How about either
```
OptionalLong scheduleAppend(int epoch, List<T> records);
```
or
```
void scheduleAppend(int epoch, List<T> records) throws BusyException;
```
I am okay with either solution but I am wondering why did you decide to return a `null` for this case instead of throwing some exception?
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java
##########
@@ -0,0 +1,296 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.Closeable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * TODO: Also flush after minimum size limit is reached?
Review comment:
I vote yes for this.
I think if we use this for writing snapshot from the state machine, then minimum size is a more interesting metrics for flushing to disk vs lingerMs.
If we implement this so that either one has to be true then the client can set the `lingerMs` or `minSize` to MAX_VALUE if it wants to ignore those values.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509620795
##########
File path: core/src/main/scala/kafka/tools/TestRaftServer.scala
##########
@@ -272,7 +291,79 @@ class TestRaftServer(val config: KafkaConfig) extends Logging {
)
}
- class RaftIoThread(client: KafkaRaftClient) extends ShutdownableThread("raft-io-thread") {
+ class RaftWorkloadGenerator(
+ client: KafkaRaftClient[Array[Byte]],
+ time: Time,
+ brokerId: Int,
+ recordsPerSec: Int,
+ recordSize: Int
+ ) extends ShutdownableThread(name = "raft-workload-generator") with RaftClient.Listener[Array[Byte]] {
+
+ private val stats = new WriteStats(time, printIntervalMs = 5000)
+ private val payload = new Array[Byte](recordSize)
+ private val pendingAppends = new util.ArrayDeque[PendingAppend]()
+
+ private var latestLeaderAndEpoch = new LeaderAndEpoch(OptionalInt.empty, 0)
+ private var isLeader = false
+ private var throttler: ThroughputThrottler = _
+ private var recordCount = 0
+
+ override def doWork(): Unit = {
+ if (latestLeaderAndEpoch != client.currentLeaderAndEpoch()) {
+ latestLeaderAndEpoch = client.currentLeaderAndEpoch()
+ isLeader = latestLeaderAndEpoch.leaderId.orElse(-1) == brokerId
+ if (isLeader) {
+ pendingAppends.clear()
+ throttler = new ThroughputThrottler(time, recordsPerSec)
+ recordCount = 0
+ }
+ }
+
+ if (isLeader) {
+ recordCount += 1
+
+ val startTimeMs = time.milliseconds()
+ val sendTimeMs = if (throttler.maybeThrottle(recordCount, startTimeMs)) {
+ time.milliseconds()
+ } else {
+ startTimeMs
+ }
+
+ val offset = client.scheduleAppend(latestLeaderAndEpoch.epoch, Collections.singletonList(payload))
+ if (offset == null || offset == Long.MaxValue) {
Review comment:
I will try to document this better, but `Long.MaxValue` is how we decided to handle the case where the epoch in `scheduleAppend` does not match the current epoch. This can happen because the raft epoch is updated asynchronously and there is no way to ensure the state machine has seen the latest value. The expectation is that the state machine will update its uncommitted state with an offset which is impossible to become committed. After it observes the epoch change, this uncommitted state will be discarded.
Note that although I added the explicit check here, it is not technically necessary. Let me consider removing it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509719785
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1605,100 +1708,18 @@ public void poll() throws IOException {
}
}
- private void failPendingAppends(KafkaException exception) {
- for (UnwrittenAppend unwrittenAppend : unwrittenAppends) {
- unwrittenAppend.fail(exception);
- }
- unwrittenAppends.clear();
- }
-
- private void pollPendingAppends(LeaderState state, long currentTimeMs) {
- int numAppends = 0;
- int maxNumAppends = unwrittenAppends.size();
-
- while (!unwrittenAppends.isEmpty() && numAppends < maxNumAppends) {
- final UnwrittenAppend unwrittenAppend = unwrittenAppends.poll();
-
- if (unwrittenAppend.future.isDone())
- continue;
-
- if (unwrittenAppend.isTimedOut(currentTimeMs)) {
- unwrittenAppend.fail(new TimeoutException("Request timeout " + unwrittenAppend.requestTimeoutMs
- + " expired before the records could be appended to the log"));
- } else {
- int epoch = quorum.epoch();
- LogAppendInfo info = appendAsLeader(unwrittenAppend.records);
- OffsetAndEpoch offsetAndEpoch = new OffsetAndEpoch(info.lastOffset, epoch);
- long numRecords = info.lastOffset - info.firstOffset + 1;
- logger.debug("Completed write of {} records at {}", numRecords, offsetAndEpoch);
-
- if (unwrittenAppend.ackMode == AckMode.LEADER) {
- unwrittenAppend.complete(offsetAndEpoch);
- } else if (unwrittenAppend.ackMode == AckMode.QUORUM) {
- CompletableFuture<Long> future = appendPurgatory.await(
- LogOffset.awaitCommitted(offsetAndEpoch.offset),
- unwrittenAppend.requestTimeoutMs);
-
- future.whenComplete((completionTimeMs, exception) -> {
- if (exception != null) {
- logger.error("Failed to commit append at {} due to {}", offsetAndEpoch, exception);
-
- unwrittenAppend.fail(exception);
- } else {
- long elapsedTime = Math.max(0, completionTimeMs - currentTimeMs);
- double elapsedTimePerRecord = (double) elapsedTime / numRecords;
- kafkaRaftMetrics.updateCommitLatency(elapsedTimePerRecord, currentTimeMs);
- unwrittenAppend.complete(offsetAndEpoch);
-
- logger.debug("Completed commit of {} records at {}", numRecords, offsetAndEpoch);
- }
- });
- }
- }
-
- numAppends++;
- }
-
- if (numAppends > 0) {
- flushLeaderLog(state, currentTimeMs);
- }
- }
-
- /**
- * Append a set of records to the log. Successful completion of the future indicates a success of
- * the append, with the uncommitted base offset and epoch.
- *
- * @param records The records to write to the log
- * @param ackMode The commit mode for the appended records
- * @param timeoutMs The maximum time to wait for the append operation to complete (including
- * any time needed for replication)
- * @return The uncommitted base offset and epoch of the appended records
- */
@Override
- public CompletableFuture<OffsetAndEpoch> append(
- Records records,
- AckMode ackMode,
- long timeoutMs
- ) {
- if (records.sizeInBytes() == 0)
- throw new IllegalArgumentException("Attempt to append empty record set");
-
- if (shutdown.get() != null)
- throw new IllegalStateException("Cannot append records while we are shutting down");
-
- if (quorum.isObserver())
- throw new IllegalStateException("Illegal attempt to write to an observer");
-
- CompletableFuture<OffsetAndEpoch> future = new CompletableFuture<>();
- UnwrittenAppend unwrittenAppend = new UnwrittenAppend(
- records, time.milliseconds(), timeoutMs, ackMode, future);
+ public Long scheduleAppend(int epoch, List<T> records) {
+ BatchAccumulator<T> accumulator = this.accumulator;
+ if (accumulator == null) {
+ return Long.MAX_VALUE;
Review comment:
Yeah, see my comment above about the handling of `Long.MAX_VALUE`. This is an attempt to reduce the error handling in the state machine. The model that we are working toward here is the following:
1) the state machine gets notified that the node has become leader in some epoch
2) the state machine can schedule appends with this epoch and it will get back the expected append offset
3) the state machine treats scheduled appends as uncommitted until the call to `handleCommit`
4) if the node resigns its leadership, the state machine will get notified and it will be expected to drop uncommitted data
By using a sort of impossible offset sentinel, the state machine just needs to wait for the notification that the leader has resigned.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r511112281
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java
##########
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.Closeable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * TODO: Also flush after minimum size limit is reached?
+ */
+public class BatchAccumulator<T> implements Closeable {
+ private final int epoch;
+ private final Time time;
+ private final Timer lingerTimer;
+ private final int lingerMs;
+ private final int maxBatchSize;
+ private final CompressionType compressionType;
+ private final MemoryPool memoryPool;
+ private final ReentrantLock lock;
+ private final RecordSerde<T> serde;
+
+ private long nextOffset;
+ private BatchBuilder<T> currentBatch;
+ private List<CompletedBatch<T>> completed;
+
+ public BatchAccumulator(
+ int epoch,
+ long baseOffset,
+ int lingerMs,
+ int maxBatchSize,
+ MemoryPool memoryPool,
+ Time time,
+ CompressionType compressionType,
+ RecordSerde<T> serde
+ ) {
+ this.epoch = epoch;
+ this.lingerMs = lingerMs;
+ this.maxBatchSize = maxBatchSize;
+ this.memoryPool = memoryPool;
+ this.time = time;
+ this.lingerTimer = time.timer(lingerMs);
+ this.compressionType = compressionType;
+ this.serde = serde;
+ this.nextOffset = baseOffset;
+ this.completed = new ArrayList<>();
+ this.lock = new ReentrantLock();
+ }
+
+ /**
+ * Append a list of records into an atomic batch. We guarantee all records
+ * are included in the same underlying record batch so that either all of
+ * the records become committed or none of them do.
+ *
+ * @param epoch the expected leader epoch
+ * @param records the list of records to include in a batch
+ * @return the offset of the last message or {@link Long#MAX_VALUE} if the epoch
+ * does not match
+ */
+ public Long append(int epoch, List<T> records) {
+ if (epoch != this.epoch) {
+ // If the epoch does not match, then the state machine probably
+ // has not gotten the notification about the latest epoch change.
+ // In this case, ignore the append and return a large offset value
+ // which will never be committed.
+ return Long.MAX_VALUE;
+ }
+
+ Object serdeContext = serde.newWriteContext();
+ int batchSize = 0;
+ for (T record : records) {
+ batchSize += serde.recordSize(record, serdeContext);
+ }
+
+ if (batchSize > maxBatchSize) {
+ throw new IllegalArgumentException("The total size of " + records + " is " + batchSize +
+ ", which exceeds the maximum allowed batch size of " + maxBatchSize);
+ }
+
+ lock.lock();
+ try {
+ BatchBuilder<T> batch = maybeAllocateBatch(batchSize);
+ if (batch == null) {
+ return null;
+ }
+
+ if (isEmpty()) {
+ lingerTimer.update();
+ lingerTimer.reset(lingerMs);
+ }
+
+ for (T record : records) {
+ batch.appendRecord(record, serdeContext);
+ nextOffset += 1;
+ }
+
+ return nextOffset - 1;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private BatchBuilder<T> maybeAllocateBatch(int batchSize) {
+ if (currentBatch == null) {
+ startNewBatch();
+ } else if (!currentBatch.hasRoomFor(batchSize)) {
+ completeCurrentBatch();
+ }
+ return currentBatch;
+ }
+
+ private void completeCurrentBatch() {
+ MemoryRecords data = currentBatch.build();
+ completed.add(new CompletedBatch<>(
+ currentBatch.baseOffset(),
+ currentBatch.records(),
+ data,
+ memoryPool,
+ currentBatch.initialBuffer()
+ ));
+ currentBatch = null;
+ startNewBatch();
+ }
+
+ private void startNewBatch() {
+ ByteBuffer buffer = memoryPool.tryAllocate(maxBatchSize);
+ if (buffer != null) {
+ currentBatch = new BatchBuilder<>(
+ buffer,
+ serde,
+ compressionType,
+ nextOffset,
+ time.milliseconds(),
+ false,
+ RecordBatch.NO_PARTITION_LEADER_EPOCH,
+ maxBatchSize
+ );
+ }
+ }
+
+ /**
+ * Check whether there are any batches which need flushing now.
Review comment:
Throughout this type we use flush to mean "collect the current batch" and start a new one. When I started reading this code I was assuming that it meant flush to disk/IO. Should we use a different verb for this? I think Java tends to use "collect" for this type of operation. What do you think?
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java
##########
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.Closeable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * TODO: Also flush after minimum size limit is reached?
+ */
+public class BatchAccumulator<T> implements Closeable {
+ private final int epoch;
+ private final Time time;
+ private final Timer lingerTimer;
+ private final int lingerMs;
+ private final int maxBatchSize;
+ private final CompressionType compressionType;
+ private final MemoryPool memoryPool;
+ private final ReentrantLock lock;
+ private final RecordSerde<T> serde;
+
+ private long nextOffset;
+ private BatchBuilder<T> currentBatch;
+ private List<CompletedBatch<T>> completed;
+
+ public BatchAccumulator(
+ int epoch,
+ long baseOffset,
+ int lingerMs,
+ int maxBatchSize,
+ MemoryPool memoryPool,
+ Time time,
+ CompressionType compressionType,
+ RecordSerde<T> serde
+ ) {
+ this.epoch = epoch;
+ this.lingerMs = lingerMs;
+ this.maxBatchSize = maxBatchSize;
+ this.memoryPool = memoryPool;
+ this.time = time;
+ this.lingerTimer = time.timer(lingerMs);
+ this.compressionType = compressionType;
+ this.serde = serde;
+ this.nextOffset = baseOffset;
+ this.completed = new ArrayList<>();
+ this.lock = new ReentrantLock();
+ }
+
+ /**
+ * Append a list of records into an atomic batch. We guarantee all records
+ * are included in the same underlying record batch so that either all of
+ * the records become committed or none of them do.
+ *
+ * @param epoch the expected leader epoch
+ * @param records the list of records to include in a batch
+ * @return the offset of the last message or {@link Long#MAX_VALUE} if the epoch
+ * does not match
+ */
+ public Long append(int epoch, List<T> records) {
+ if (epoch != this.epoch) {
+ // If the epoch does not match, then the state machine probably
+ // has not gotten the notification about the latest epoch change.
+ // In this case, ignore the append and return a large offset value
+ // which will never be committed.
Review comment:
What are you trying to protect with this check? For example, the signature could be `public long append(List<T> records)` with the accumulator writing the correct epoch.
In https://github.com/apache/kafka/pull/9482 you implemented `handleClaim` in to only fire when the `Listener`'s "acknowledged" offset + 1 is >= to the leader's epoch start offset.
Thinking through the code's behaviour, I see this check catching the case the the raft replica lost leadership and won leadership before the `Listener` was able to asynchronously process `handleResign` and `handleClaim`.
##########
File path: core/src/main/scala/kafka/raft/KafkaNetworkChannel.scala
##########
@@ -216,11 +216,9 @@ class KafkaNetworkChannel(time: Time,
endpoints.put(id, node)
}
- def postInboundRequest(header: RequestHeader,
- request: AbstractRequest,
- onResponseReceived: ResponseHandler): Unit = {
+ def postInboundRequest(request: AbstractRequest, onResponseReceived: ResponseHandler): Unit = {
Review comment:
Currently, `postInboundRequest` in only used by `TestRaftRequestHandler` and `KafkaNetworkChannelTest`. Are you thinking that we will use the same or similar method when integrating with the broker code?
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchMemoryPool.java
##########
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.memory.MemoryPool;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Simple memory pool which maintains a limited number of fixed-size buffers.
+ */
+public class BatchMemoryPool implements MemoryPool {
+ private final ReentrantLock lock;
+ private final Deque<ByteBuffer> free;
+ private final int maxBatches;
+ private final int batchSize;
+
+ private int numAllocatedBatches = 0;
+
+ public BatchMemoryPool(int maxBatches, int batchSize) {
+ this.maxBatches = maxBatches;
+ this.batchSize = batchSize;
+ this.free = new ArrayDeque<>(maxBatches);
+ this.lock = new ReentrantLock();
+ }
+
+ @Override
+ public ByteBuffer tryAllocate(int sizeBytes) {
+ if (sizeBytes > batchSize) {
+ throw new IllegalArgumentException("Cannot allocate buffers larger than max " +
+ "batch size of " + batchSize);
+ }
+
+ lock.lock();
+ try {
+ ByteBuffer buffer = free.poll();
+ if (buffer == null && numAllocatedBatches < maxBatches) {
+ buffer = ByteBuffer.allocate(batchSize);
+ numAllocatedBatches += 1;
+ }
+
+ if (buffer != null) {
+ buffer.clear();
+ buffer.limit(sizeBytes);
+ }
+ return buffer;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void release(ByteBuffer previouslyAllocated) {
Review comment:
Let's check that `previouslyAllocated`'s capacity is `batchSize` else `buffer.limit(sizeBytes)` is going to throw with a less useful stacktrace.
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchBuilder.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.protocol.DataOutputStreamWritable;
+import org.apache.kafka.common.record.AbstractRecords;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.DefaultRecord;
+import org.apache.kafka.common.record.DefaultRecordBatch;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.DataOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class BatchBuilder<T> {
Review comment:
I think it would be good to have a description of what this type is doing and how to use it. Reading the code and seeing how the buffer is shared, I am under the impression that this type should not be reused after the `build` method is called. Is that correct?
If this is correct, I think all of the public methods should check that they are not being called after `build` is called.
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java
##########
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.Closeable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * TODO: Also flush after minimum size limit is reached?
+ */
+public class BatchAccumulator<T> implements Closeable {
+ private final int epoch;
+ private final Time time;
+ private final Timer lingerTimer;
+ private final int lingerMs;
+ private final int maxBatchSize;
+ private final CompressionType compressionType;
+ private final MemoryPool memoryPool;
+ private final ReentrantLock lock;
+ private final RecordSerde<T> serde;
+
+ private long nextOffset;
+ private BatchBuilder<T> currentBatch;
+ private List<CompletedBatch<T>> completed;
+
+ public BatchAccumulator(
+ int epoch,
+ long baseOffset,
+ int lingerMs,
+ int maxBatchSize,
+ MemoryPool memoryPool,
+ Time time,
+ CompressionType compressionType,
+ RecordSerde<T> serde
+ ) {
+ this.epoch = epoch;
+ this.lingerMs = lingerMs;
+ this.maxBatchSize = maxBatchSize;
+ this.memoryPool = memoryPool;
+ this.time = time;
+ this.lingerTimer = time.timer(lingerMs);
+ this.compressionType = compressionType;
+ this.serde = serde;
+ this.nextOffset = baseOffset;
+ this.completed = new ArrayList<>();
+ this.lock = new ReentrantLock();
+ }
+
+ /**
+ * Append a list of records into an atomic batch. We guarantee all records
+ * are included in the same underlying record batch so that either all of
+ * the records become committed or none of them do.
+ *
+ * @param epoch the expected leader epoch
+ * @param records the list of records to include in a batch
+ * @return the offset of the last message or {@link Long#MAX_VALUE} if the epoch
+ * does not match
Review comment:
We should document this returns `null` and what it means when `null` is returned.
##########
File path: core/src/main/scala/kafka/raft/KafkaNetworkChannel.scala
##########
@@ -216,11 +216,9 @@ class KafkaNetworkChannel(time: Time,
endpoints.put(id, node)
}
- def postInboundRequest(header: RequestHeader,
- request: AbstractRequest,
- onResponseReceived: ResponseHandler): Unit = {
+ def postInboundRequest(request: AbstractRequest, onResponseReceived: ResponseHandler): Unit = {
Review comment:
Unrelated with this change but when reading the code. I was confused by the name `pollInboundResponses`. That function returns both pending request and responses.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r512390308
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchBuilder.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.protocol.DataOutputStreamWritable;
+import org.apache.kafka.common.record.AbstractRecords;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.DefaultRecord;
+import org.apache.kafka.common.record.DefaultRecordBatch;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.DataOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class BatchBuilder<T> {
Review comment:
I think the pairs of those classes:
1) BatchAccumulator and RecordAccumulator
2) BatchMemoryPool and BufferPool
3) BatchBuilder and MemoryRecordsBuilder
are similar can maybe can be consolidated to a single class each.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r512389580
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1605,100 +1708,18 @@ public void poll() throws IOException {
}
}
- private void failPendingAppends(KafkaException exception) {
- for (UnwrittenAppend unwrittenAppend : unwrittenAppends) {
- unwrittenAppend.fail(exception);
- }
- unwrittenAppends.clear();
- }
-
- private void pollPendingAppends(LeaderState state, long currentTimeMs) {
- int numAppends = 0;
- int maxNumAppends = unwrittenAppends.size();
-
- while (!unwrittenAppends.isEmpty() && numAppends < maxNumAppends) {
- final UnwrittenAppend unwrittenAppend = unwrittenAppends.poll();
-
- if (unwrittenAppend.future.isDone())
- continue;
-
- if (unwrittenAppend.isTimedOut(currentTimeMs)) {
- unwrittenAppend.fail(new TimeoutException("Request timeout " + unwrittenAppend.requestTimeoutMs
- + " expired before the records could be appended to the log"));
- } else {
- int epoch = quorum.epoch();
- LogAppendInfo info = appendAsLeader(unwrittenAppend.records);
- OffsetAndEpoch offsetAndEpoch = new OffsetAndEpoch(info.lastOffset, epoch);
- long numRecords = info.lastOffset - info.firstOffset + 1;
- logger.debug("Completed write of {} records at {}", numRecords, offsetAndEpoch);
-
- if (unwrittenAppend.ackMode == AckMode.LEADER) {
- unwrittenAppend.complete(offsetAndEpoch);
- } else if (unwrittenAppend.ackMode == AckMode.QUORUM) {
- CompletableFuture<Long> future = appendPurgatory.await(
- LogOffset.awaitCommitted(offsetAndEpoch.offset),
- unwrittenAppend.requestTimeoutMs);
-
- future.whenComplete((completionTimeMs, exception) -> {
- if (exception != null) {
- logger.error("Failed to commit append at {} due to {}", offsetAndEpoch, exception);
-
- unwrittenAppend.fail(exception);
- } else {
- long elapsedTime = Math.max(0, completionTimeMs - currentTimeMs);
- double elapsedTimePerRecord = (double) elapsedTime / numRecords;
- kafkaRaftMetrics.updateCommitLatency(elapsedTimePerRecord, currentTimeMs);
- unwrittenAppend.complete(offsetAndEpoch);
-
- logger.debug("Completed commit of {} records at {}", numRecords, offsetAndEpoch);
- }
- });
- }
- }
-
- numAppends++;
- }
-
- if (numAppends > 0) {
- flushLeaderLog(state, currentTimeMs);
- }
- }
-
- /**
- * Append a set of records to the log. Successful completion of the future indicates a success of
- * the append, with the uncommitted base offset and epoch.
- *
- * @param records The records to write to the log
- * @param ackMode The commit mode for the appended records
- * @param timeoutMs The maximum time to wait for the append operation to complete (including
- * any time needed for replication)
- * @return The uncommitted base offset and epoch of the appended records
- */
@Override
- public CompletableFuture<OffsetAndEpoch> append(
- Records records,
- AckMode ackMode,
- long timeoutMs
- ) {
- if (records.sizeInBytes() == 0)
- throw new IllegalArgumentException("Attempt to append empty record set");
-
- if (shutdown.get() != null)
- throw new IllegalStateException("Cannot append records while we are shutting down");
-
- if (quorum.isObserver())
- throw new IllegalStateException("Illegal attempt to write to an observer");
-
- CompletableFuture<OffsetAndEpoch> future = new CompletableFuture<>();
- UnwrittenAppend unwrittenAppend = new UnwrittenAppend(
- records, time.milliseconds(), timeoutMs, ackMode, future);
+ public Long scheduleAppend(int epoch, List<T> records) {
+ BatchAccumulator<T> accumulator = this.accumulator;
+ if (accumulator == null) {
+ return Long.MAX_VALUE;
Review comment:
Sounds good.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r512195616
##########
File path: core/src/main/scala/kafka/raft/KafkaNetworkChannel.scala
##########
@@ -216,11 +216,9 @@ class KafkaNetworkChannel(time: Time,
endpoints.put(id, node)
}
- def postInboundRequest(header: RequestHeader,
- request: AbstractRequest,
- onResponseReceived: ResponseHandler): Unit = {
+ def postInboundRequest(request: AbstractRequest, onResponseReceived: ResponseHandler): Unit = {
Review comment:
> I don't think that's right. It only delivers responses for outbound requests (in other words, inbound responses).
Hmm. I misspoke a bit. I should have said that `pollInboundResponses` returns both inbound requests and responses. These are my observations:
`postInboundRequest` puts requests in `undelivered`:
https://github.com/apache/kafka/pull/9418/files#diff-c9b2a05e69be1d2391d91a33f408eab4fde1245d4ea9cf9f52305c8cb89208ceR224
`pollInboundResponses` returns everything in `undelivered`:
https://github.com/apache/kafka/pull/9418/files#diff-c9b2a05e69be1d2391d91a33f408eab4fde1245d4ea9cf9f52305c8cb89208ceR201-R202
The raft client gets all of the inbound messages (requests and reponses) by calling `receive` which in term calls `pollInboundResponses`:
https://github.com/apache/kafka/pull/9418/files#diff-1da15c51e641ea46ea5c86201ab8f21cfee9e7c575102a39c7bae0d5ffd7de39L1596
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509726250
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchBuilder.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.protocol.DataOutputStreamWritable;
+import org.apache.kafka.common.record.AbstractRecords;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.DefaultRecord;
+import org.apache.kafka.common.record.DefaultRecordBatch;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.DataOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class BatchBuilder<T> {
Review comment:
I am not sure I follow. Are you suggesting making `BatchBuilder` a nested class? I think what it's doing is complex enough that I wanted a separate class that could be tested in isolation.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r511200238
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java
##########
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.Closeable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * TODO: Also flush after minimum size limit is reached?
+ */
+public class BatchAccumulator<T> implements Closeable {
+ private final int epoch;
+ private final Time time;
+ private final Timer lingerTimer;
+ private final int lingerMs;
+ private final int maxBatchSize;
+ private final CompressionType compressionType;
+ private final MemoryPool memoryPool;
+ private final ReentrantLock lock;
+ private final RecordSerde<T> serde;
+
+ private long nextOffset;
+ private BatchBuilder<T> currentBatch;
+ private List<CompletedBatch<T>> completed;
+
+ public BatchAccumulator(
+ int epoch,
+ long baseOffset,
+ int lingerMs,
+ int maxBatchSize,
+ MemoryPool memoryPool,
+ Time time,
+ CompressionType compressionType,
+ RecordSerde<T> serde
+ ) {
+ this.epoch = epoch;
+ this.lingerMs = lingerMs;
+ this.maxBatchSize = maxBatchSize;
+ this.memoryPool = memoryPool;
+ this.time = time;
+ this.lingerTimer = time.timer(lingerMs);
+ this.compressionType = compressionType;
+ this.serde = serde;
+ this.nextOffset = baseOffset;
+ this.completed = new ArrayList<>();
+ this.lock = new ReentrantLock();
+ }
+
+ /**
+ * Append a list of records into an atomic batch. We guarantee all records
+ * are included in the same underlying record batch so that either all of
+ * the records become committed or none of them do.
+ *
+ * @param epoch the expected leader epoch
+ * @param records the list of records to include in a batch
+ * @return the offset of the last message or {@link Long#MAX_VALUE} if the epoch
+ * does not match
+ */
+ public Long append(int epoch, List<T> records) {
+ if (epoch != this.epoch) {
+ // If the epoch does not match, then the state machine probably
+ // has not gotten the notification about the latest epoch change.
+ // In this case, ignore the append and return a large offset value
+ // which will never be committed.
Review comment:
Right. It is important to ensure that the state machine has observed the latest leader epoch. Otherwise there may be committed data inflight which the state machine has yet to see.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509603110
##########
File path: clients/src/main/java/org/apache/kafka/common/protocol/DataOutputWritable.java
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.protocol;
+
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.common.utils.Utils;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class DataOutputWritable implements Writable {
Review comment:
Fair enough. We have no current need for `DataOutputWritable`, so I will remove this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509633527
##########
File path: core/src/main/scala/kafka/tools/TestRaftServer.scala
##########
@@ -272,7 +291,79 @@ class TestRaftServer(val config: KafkaConfig) extends Logging {
)
}
- class RaftIoThread(client: KafkaRaftClient) extends ShutdownableThread("raft-io-thread") {
+ class RaftWorkloadGenerator(
+ client: KafkaRaftClient[Array[Byte]],
+ time: Time,
+ brokerId: Int,
+ recordsPerSec: Int,
+ recordSize: Int
+ ) extends ShutdownableThread(name = "raft-workload-generator") with RaftClient.Listener[Array[Byte]] {
+
+ private val stats = new WriteStats(time, printIntervalMs = 5000)
+ private val payload = new Array[Byte](recordSize)
+ private val pendingAppends = new util.ArrayDeque[PendingAppend]()
+
+ private var latestLeaderAndEpoch = new LeaderAndEpoch(OptionalInt.empty, 0)
+ private var isLeader = false
+ private var throttler: ThroughputThrottler = _
+ private var recordCount = 0
+
+ override def doWork(): Unit = {
+ if (latestLeaderAndEpoch != client.currentLeaderAndEpoch()) {
+ latestLeaderAndEpoch = client.currentLeaderAndEpoch()
+ isLeader = latestLeaderAndEpoch.leaderId.orElse(-1) == brokerId
+ if (isLeader) {
+ pendingAppends.clear()
+ throttler = new ThroughputThrottler(time, recordsPerSec)
+ recordCount = 0
+ }
+ }
+
+ if (isLeader) {
+ recordCount += 1
+
+ val startTimeMs = time.milliseconds()
+ val sendTimeMs = if (throttler.maybeThrottle(recordCount, startTimeMs)) {
+ time.milliseconds()
+ } else {
+ startTimeMs
+ }
+
+ val offset = client.scheduleAppend(latestLeaderAndEpoch.epoch, Collections.singletonList(payload))
+ if (offset == null || offset == Long.MaxValue) {
+ time.sleep(10)
+ } else {
+ pendingAppends.offer(PendingAppend(latestLeaderAndEpoch.epoch, offset, sendTimeMs))
+ }
+ } else {
+ time.sleep(500)
+ }
+ }
+
+ override def handleCommit(epoch: Int, lastOffset: Long, records: util.List[Array[Byte]]): Unit = {
+ var offset = lastOffset - records.size() + 1
+ val currentTimeMs = time.milliseconds()
+
+ for (record <- records.asScala) {
+ val pendingAppend = pendingAppends.poll()
+ if (pendingAppend.epoch != epoch || pendingAppend.offset!= offset) {
+ warn(s"Expected next commit at offset ${pendingAppend.offset}, " +
Review comment:
That's true. In a follow-up, the `handleCommit` API will be expanded a bit to cover appends through replication as well, but for now, I think we can raise an error.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509723127
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java
##########
@@ -0,0 +1,296 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.Closeable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * TODO: Also flush after minimum size limit is reached?
+ */
+public class BatchAccumulator<T> implements Closeable {
+ private final int epoch;
+ private final Time time;
+ private final Timer lingerTimer;
+ private final int lingerMs;
+ private final int maxBatchSize;
+ private final CompressionType compressionType;
+ private final MemoryPool memoryPool;
+ private final ReentrantLock lock;
+ private final RecordSerde<T> serde;
+
+ private long nextOffset;
+ private BatchBuilder<T> currentBatch;
+ private List<CompletedBatch<T>> completed;
+
+ public BatchAccumulator(
+ int epoch,
+ long baseOffset,
+ int lingerMs,
+ int maxBatchSize,
+ MemoryPool memoryPool,
+ Time time,
+ CompressionType compressionType,
+ RecordSerde<T> serde
+ ) {
+ this.epoch = epoch;
+ this.lingerMs = lingerMs;
+ this.maxBatchSize = maxBatchSize;
+ this.memoryPool = memoryPool;
+ this.time = time;
+ this.lingerTimer = time.timer(lingerMs);
+ this.compressionType = compressionType;
+ this.serde = serde;
+ this.nextOffset = baseOffset;
+ this.completed = new ArrayList<>();
+ this.lock = new ReentrantLock();
+ }
+
+ /**
+ * Append a list of records into an atomic batch. We guarantee all records
+ * are included in the same underlying record batch so that either all of
+ * the records become committed or none of them do.
+ *
+ * @param epoch the expected leader epoch
+ * @param records the list of records to include in a batch
+ * @return the offset of the last message or {@link Long#MAX_VALUE} if the epoch
+ * does not match
+ */
+ public Long append(int epoch, List<T> records) {
+ if (epoch != this.epoch) {
+ // If the epoch does not match, then the state machine probably
+ // has not gotten the notification about the latest epoch change.
+ // In this case, ignore the append and return a large offset value
+ // which will never be committed.
+ return Long.MAX_VALUE;
+ }
+
+ Object serdeContext = serde.newWriteContext();
+ int batchSize = 0;
+ for (T record : records) {
+ batchSize += serde.recordSize(record, serdeContext);
+ }
+
+ if (batchSize > maxBatchSize) {
+ throw new IllegalArgumentException("The total size of " + records + " is " + batchSize +
+ ", which exceeds the maximum allowed batch size of " + maxBatchSize);
+ }
+
+ lock.lock();
+ try {
+ BatchBuilder<T> batch = maybeAllocateBatch(batchSize);
+ if (batch == null) {
+ return null;
+ }
+
+ if (isEmpty()) {
+ lingerTimer.update();
+ lingerTimer.reset(lingerMs);
+ }
+
+ for (T record : records) {
+ batch.appendRecord(record, serdeContext);
+ nextOffset += 1;
+ }
+
+ return nextOffset - 1;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private BatchBuilder<T> maybeAllocateBatch(int batchSize) {
+ if (currentBatch == null) {
+ startNewBatch();
+ } else if (!currentBatch.hasRoomFor(batchSize)) {
+ completeCurrentBatch();
+ }
+ return currentBatch;
+ }
+
+ private void completeCurrentBatch() {
+ MemoryRecords data = currentBatch.build();
+ completed.add(new CompletedBatch<>(
+ currentBatch.baseOffset(),
+ currentBatch.records(),
+ data,
+ memoryPool,
+ currentBatch.initialBuffer()
+ ));
+ currentBatch = null;
+ startNewBatch();
+ }
+
+ private void startNewBatch() {
+ ByteBuffer buffer = memoryPool.tryAllocate(maxBatchSize);
+ if (buffer != null) {
+ currentBatch = new BatchBuilder<>(
+ buffer,
+ serde,
+ compressionType,
+ nextOffset,
+ time.milliseconds(),
+ false,
+ RecordBatch.NO_PARTITION_LEADER_EPOCH,
+ maxBatchSize
+ );
+ }
+ }
+
+ /**
+ * Check whether there are any batches which need flushing now.
+ *
+ * @param currentTimeMs current time in milliseconds
+ * @return true if there are batches ready to flush, false otherwise
+ */
+ public boolean needsFlush(long currentTimeMs) {
+ return timeUntilFlush(currentTimeMs) <= 0;
+ }
+
+ /**
+ * Check the time remaining until the next needed flush. If the accumulator
+ * is empty, then {@link Long#MAX_VALUE} will be returned.
+ *
+ * @param currentTimeMs current time in milliseconds
+ * @return the delay in milliseconds before the next expected flush
+ */
+ public long timeUntilFlush(long currentTimeMs) {
+ lock.lock();
+ try {
+ lingerTimer.update(currentTimeMs);
+ if (isEmpty()) {
+ return Long.MAX_VALUE;
+ } else {
+ return lingerTimer.remainingMs();
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private boolean isEmpty() {
+ lock.lock();
+ try {
+ if (currentBatch != null && currentBatch.nonEmpty()) {
+ return false;
+ } else {
+ return completed.isEmpty();
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Get the leader epoch, which is constant for each instance.
+ *
+ * @return the leader epoch
+ */
+ public int epoch() {
+ return epoch;
+ }
+
+ /**
+ * Flush completed batches. The caller is expected to first check whether
+ * a flush is expected using {@link #needsFlush(long)} in order to avoid
+ * unnecessary flushing.
+ *
+ * @return the list of completed batches
+ */
+ public List<CompletedBatch<T>> flush() {
+ lock.lock();
+ try {
+ if (currentBatch != null && currentBatch.nonEmpty()) {
+ completeCurrentBatch();
+ }
+
+ List<CompletedBatch<T>> res = completed;
+ this.completed = new ArrayList<>();
+ return res;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Get the number of batches including the one that is currently being
+ * written to (if it exists).
+ *
+ * @return
+ */
+ public int count() {
Review comment:
I am not sure. We don't have a metric yet, so I thought we might as well start with a thread-safe implementation. If we add a metric in the future, we can probably use an `AtomicInteger` or something and eliminate the locking.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r509636357
##########
File path: core/src/main/scala/kafka/tools/TestRaftServer.scala
##########
@@ -19,37 +19,44 @@ package kafka.tools
import java.io.File
import java.nio.file.Files
-import java.util.concurrent.CountDownLatch
-import java.util.{Properties, Random}
+import java.util
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+import java.util.{Collections, OptionalInt, Random}
-import joptsimple.OptionParser
+import com.yammer.metrics.core.MetricName
+import joptsimple.OptionException
import kafka.log.{Log, LogConfig, LogManager}
import kafka.network.SocketServer
import kafka.raft.{KafkaFuturePurgatory, KafkaMetadataLog, KafkaNetworkChannel}
import kafka.security.CredentialProvider
import kafka.server.{BrokerTopicStats, KafkaConfig, KafkaRequestHandlerPool, KafkaServer, LogDirFailureChannel}
+import kafka.tools.TestRaftServer.{ByteArraySerde, PendingAppend, ThroughputThrottler, WriteStats}
Review comment:
Moved the import inside `TestRaftServer`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r512224767
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchBuilder.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.protocol.DataOutputStreamWritable;
+import org.apache.kafka.common.record.AbstractRecords;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.DefaultRecord;
+import org.apache.kafka.common.record.DefaultRecordBatch;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+import java.io.DataOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class BatchBuilder<T> {
Review comment:
Sure, we can do that. I was being a bit looser with API validation since this is bound for internal use only.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r512196964
##########
File path: core/src/main/scala/kafka/raft/KafkaNetworkChannel.scala
##########
@@ -216,11 +216,9 @@ class KafkaNetworkChannel(time: Time,
endpoints.put(id, node)
}
- def postInboundRequest(header: RequestHeader,
- request: AbstractRequest,
- onResponseReceived: ResponseHandler): Unit = {
+ def postInboundRequest(request: AbstractRequest, onResponseReceived: ResponseHandler): Unit = {
Review comment:
My mistake. This _was_ being used for both inbound requests and responses. Let me change the name and add some docs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9418: KAFKA-10601; Add support for append linger to Raft implementation
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9418:
URL: https://github.com/apache/kafka/pull/9418#discussion_r512195616
##########
File path: core/src/main/scala/kafka/raft/KafkaNetworkChannel.scala
##########
@@ -216,11 +216,9 @@ class KafkaNetworkChannel(time: Time,
endpoints.put(id, node)
}
- def postInboundRequest(header: RequestHeader,
- request: AbstractRequest,
- onResponseReceived: ResponseHandler): Unit = {
+ def postInboundRequest(request: AbstractRequest, onResponseReceived: ResponseHandler): Unit = {
Review comment:
> I don't think that's right. It only delivers responses for outbound requests (in other words, inbound responses).
Hmm. I misspoke a bit. I should have said that `pollInboundResponses` returns both inbound requests and responses. These are my observations:
`postInboundRequest` puts requests in `undelivered`:
https://github.com/apache/kafka/pull/9418/files#diff-c9b2a05e69be1d2391d91a33f408eab4fde1245d4ea9cf9f52305c8cb89208ceR224
`pollInboundResponses` returns everything in `undelivered`:
https://github.com/apache/kafka/pull/9418/files#diff-c9b2a05e69be1d2391d91a33f408eab4fde1245d4ea9cf9f52305c8cb89208ceR201-R202
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org