You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2023/12/21 14:17:41 UTC
(kafka) branch trunk updated: KAFKA-16030: new group coordinator should check if partition goes offline during load (#15043)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 46132860765 KAFKA-16030: new group coordinator should check if partition goes offline during load (#15043)
46132860765 is described below
commit 461328607655b46d6017b07135a1d15634192a69
Author: Jeff Kim <ki...@gmail.com>
AuthorDate: Thu Dec 21 09:17:35 2023 -0500
KAFKA-16030: new group coordinator should check if partition goes offline during load (#15043)
The new coordinator stops loading if the partition goes offline during load. However, the partition is still considered active. Instead, we should return NOT_LEADER_OR_FOLLOWER exception during load.
Another change is that we only want to invoke CoordinatorPlayback#updateLastCommittedOffset if the current offset (last written offset) is greater than or equal to the current high watermark. This is to ensure that in the case the high watermark is ahead of the current offset, we don't clear snapshots prematurely.
Reviewers: David Jacot <dj...@confluent.io>
---
.../coordinator/group/CoordinatorLoaderImpl.scala | 15 ++-
.../group/CoordinatorLoaderImplTest.scala | 124 +++++++++++++++++++++
.../group/runtime/SnapshottableCoordinator.java | 5 +
.../runtime/SnapshottableCoordinatorTest.java | 1 +
4 files changed, 140 insertions(+), 5 deletions(-)
diff --git a/core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala b/core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala
index 27e555fad5a..c583e693224 100644
--- a/core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala
+++ b/core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala
@@ -160,18 +160,23 @@ class CoordinatorLoaderImpl[T](
val currentHighWatermark = log.highWatermark
if (currentOffset >= currentHighWatermark) {
coordinator.updateLastWrittenOffset(currentOffset)
- }
- if (currentHighWatermark > previousHighWatermark) {
- coordinator.updateLastCommittedOffset(currentHighWatermark)
- previousHighWatermark = currentHighWatermark
+ if (currentHighWatermark > previousHighWatermark) {
+ coordinator.updateLastCommittedOffset(currentHighWatermark)
+ previousHighWatermark = currentHighWatermark
+ }
}
}
numBytes = numBytes + memoryRecords.sizeInBytes()
}
+
val endTimeMs = time.milliseconds()
- if (isRunning.get) {
+ if (logEndOffset == -1L) {
+ future.completeExceptionally(new NotLeaderOrFollowerException(
+ s"Stopped loading records from $tp because the partition is not online or is no longer the leader."
+ ))
+ } else if (isRunning.get) {
future.complete(new LoadSummary(startTimeMs, endTimeMs, numRecords, numBytes))
} else {
future.completeExceptionally(new RuntimeException("Coordinator loader is closed."))
diff --git a/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorLoaderImplTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorLoaderImplTest.scala
index 1475e715c08..9fc10299735 100644
--- a/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorLoaderImplTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorLoaderImplTest.scala
@@ -472,6 +472,130 @@ class CoordinatorLoaderImplTest {
}
}
+ @Test
+ def testUpdateLastWrittenOffsetOnBatchLoadedWhileHighWatermarkAhead(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val serde = new StringKeyValueDeserializer
+ val log = mock(classOf[UnifiedLog])
+ val coordinator = mock(classOf[CoordinatorPlayback[(String, String)]])
+
+ TestUtils.resource(new CoordinatorLoaderImpl[(String, String)](
+ time = Time.SYSTEM,
+ replicaManager = replicaManager,
+ deserializer = serde,
+ loadBufferSize = 1000
+ )) { loader =>
+ when(replicaManager.getLog(tp)).thenReturn(Some(log))
+ when(log.logStartOffset).thenReturn(0L)
+ when(log.highWatermark).thenReturn(5L).thenReturn(7L).thenReturn(7L)
+ when(replicaManager.getLogEndOffset(tp)).thenReturn(Some(7L))
+
+ val readResult1 = logReadResult(startOffset = 0, records = Seq(
+ new SimpleRecord("k1".getBytes, "v1".getBytes),
+ new SimpleRecord("k2".getBytes, "v2".getBytes)
+ ))
+
+ when(log.read(
+ startOffset = 0L,
+ maxLength = 1000,
+ isolation = FetchIsolation.LOG_END,
+ minOneMessage = true
+ )).thenReturn(readResult1)
+
+ val readResult2 = logReadResult(startOffset = 2, records = Seq(
+ new SimpleRecord("k3".getBytes, "v3".getBytes),
+ new SimpleRecord("k4".getBytes, "v4".getBytes),
+ new SimpleRecord("k5".getBytes, "v5".getBytes)
+ ))
+
+ when(log.read(
+ startOffset = 2L,
+ maxLength = 1000,
+ isolation = FetchIsolation.LOG_END,
+ minOneMessage = true
+ )).thenReturn(readResult2)
+
+ val readResult3 = logReadResult(startOffset = 5, records = Seq(
+ new SimpleRecord("k6".getBytes, "v6".getBytes),
+ new SimpleRecord("k7".getBytes, "v7".getBytes)
+ ))
+
+ when(log.read(
+ startOffset = 5L,
+ maxLength = 1000,
+ isolation = FetchIsolation.LOG_END,
+ minOneMessage = true
+ )).thenReturn(readResult3)
+
+ assertNotNull(loader.load(tp, coordinator).get(10, TimeUnit.SECONDS))
+
+ verify(coordinator).replay(RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, ("k1", "v1"))
+ verify(coordinator).replay(RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, ("k2", "v2"))
+ verify(coordinator).replay(RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, ("k3", "v3"))
+ verify(coordinator).replay(RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, ("k4", "v4"))
+ verify(coordinator).replay(RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, ("k5", "v5"))
+ verify(coordinator).replay(RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, ("k6", "v6"))
+ verify(coordinator).replay(RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, ("k7", "v7"))
+ verify(coordinator, times(0)).updateLastWrittenOffset(0)
+ verify(coordinator, times(0)).updateLastWrittenOffset(2)
+ verify(coordinator, times(0)).updateLastWrittenOffset(5)
+ verify(coordinator, times(1)).updateLastWrittenOffset(7)
+ verify(coordinator, times(0)).updateLastCommittedOffset(0)
+ verify(coordinator, times(0)).updateLastCommittedOffset(2)
+ verify(coordinator, times(0)).updateLastCommittedOffset(5)
+ verify(coordinator, times(1)).updateLastCommittedOffset(7)
+ }
+ }
+
+ @Test
+ def testPartitionGoesOfflineDuringLoad(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val serde = new StringKeyValueDeserializer
+ val log = mock(classOf[UnifiedLog])
+ val coordinator = mock(classOf[CoordinatorPlayback[(String, String)]])
+
+ TestUtils.resource(new CoordinatorLoaderImpl[(String, String)](
+ time = Time.SYSTEM,
+ replicaManager = replicaManager,
+ deserializer = serde,
+ loadBufferSize = 1000
+ )) { loader =>
+ when(replicaManager.getLog(tp)).thenReturn(Some(log))
+ when(log.logStartOffset).thenReturn(0L)
+ when(log.highWatermark).thenReturn(0L)
+ when(replicaManager.getLogEndOffset(tp)).thenReturn(Some(5L)).thenReturn(Some(-1L))
+
+ val readResult1 = logReadResult(startOffset = 0, records = Seq(
+ new SimpleRecord("k1".getBytes, "v1".getBytes),
+ new SimpleRecord("k2".getBytes, "v2".getBytes)
+ ))
+
+ when(log.read(
+ startOffset = 0L,
+ maxLength = 1000,
+ isolation = FetchIsolation.LOG_END,
+ minOneMessage = true
+ )).thenReturn(readResult1)
+
+ val readResult2 = logReadResult(startOffset = 2, records = Seq(
+ new SimpleRecord("k3".getBytes, "v3".getBytes),
+ new SimpleRecord("k4".getBytes, "v4".getBytes),
+ new SimpleRecord("k5".getBytes, "v5".getBytes)
+ ))
+
+ when(log.read(
+ startOffset = 2L,
+ maxLength = 1000,
+ isolation = FetchIsolation.LOG_END,
+ minOneMessage = true
+ )).thenReturn(readResult2)
+
+ assertFutureThrows(loader.load(tp, coordinator), classOf[NotLeaderOrFollowerException])
+ }
+ }
+
private def logReadResult(
startOffset: Long,
producerId: Long = RecordBatch.NO_PRODUCER_ID,
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinator.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinator.java
index fdc1aa47a09..63513465569 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinator.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinator.java
@@ -143,6 +143,11 @@ class SnapshottableCoordinator<S extends CoordinatorShard<U>, U> implements Coor
" must be greater than or equal to " + lastCommittedOffset + ".");
}
+ if (offset > lastWrittenOffset) {
+ throw new IllegalStateException("New committed offset " + offset + " of " + tp +
+ "must be less than or equal to " + lastWrittenOffset + ".");
+ }
+
lastCommittedOffset = offset;
snapshotRegistry.deleteSnapshotsUpTo(offset);
log.debug("Updated committed offset of {} to {}.", tp, offset);
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinatorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinatorTest.java
index 220442c057d..0c949049b4e 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinatorTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinatorTest.java
@@ -141,5 +141,6 @@ public class SnapshottableCoordinatorTest {
assertEquals(100L, coordinator.lastCommittedOffset());
assertThrows(IllegalStateException.class, () -> coordinator.updateLastCommittedOffset(99L));
assertEquals(100L, coordinator.lastCommittedOffset());
+ assertThrows(IllegalStateException.class, () -> coordinator.updateLastCommittedOffset(101L));
}
}