You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2019/10/17 17:10:30 UTC
[kafka] branch 2.4 updated (45fd476 -> f357fc0)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a change to branch 2.4
in repository https://gitbox.apache.org/repos/asf/kafka.git.
from 45fd476 MINOR: Upgrade zk to 3.5.6 (#7544)
new 0fbf012 KAFKA-9004; Prevent older clients from fetching from a follower (#7531)
new f357fc0 MINOR: Augment log4j to add generation number in performAssign (#7451)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../consumer/internals/AbstractCoordinator.java | 66 +++++++++------
.../consumer/internals/ConsumerCoordinator.java | 11 +--
.../internals/AbstractCoordinatorTest.java | 2 +-
.../internals/ConsumerCoordinatorTest.java | 6 +-
.../runtime/distributed/WorkerCoordinator.java | 2 +-
.../main/scala/kafka/server/ReplicaManager.scala | 12 +--
.../unit/kafka/server/ReplicaManagerTest.scala | 94 +++++++++++++++++++++-
7 files changed, 150 insertions(+), 43 deletions(-)
[kafka] 02/02: MINOR: Augment log4j to add generation number in
performAssign (#7451)
Posted by jg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch 2.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit f357fc0339c11d9412f433df1920b509c575c31b
Author: Guozhang Wang <wa...@gmail.com>
AuthorDate: Wed Oct 9 10:34:19 2019 -0700
MINOR: Augment log4j to add generation number in performAssign (#7451)
Since generation is private in AbstractCoordinator, I need to modify the generation() to let it return the object directly.
Reviewers: A. Sophie Blee-Goldman <so...@confluent.io>, Bill Bejeck <bi...@confluent.io>
---
.../consumer/internals/AbstractCoordinator.java | 66 ++++++++++++++--------
.../consumer/internals/ConsumerCoordinator.java | 11 ++--
.../internals/AbstractCoordinatorTest.java | 2 +-
.../internals/ConsumerCoordinatorTest.java | 6 +-
.../runtime/distributed/WorkerCoordinator.java | 2 +-
5 files changed, 52 insertions(+), 35 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
index 057cfb0..3a76276 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -638,7 +638,7 @@ public abstract class AbstractCoordinator implements Closeable {
.setGenerationId(generation.generationId)
.setAssignments(Collections.emptyList())
);
- log.debug("Sending follower SyncGroup to coordinator {}: {}", this.coordinator, requestBuilder);
+ log.debug("Sending follower SyncGroup to coordinator {} at generation {}: {}", this.coordinator, this.generation, requestBuilder);
return sendSyncGroupRequest(requestBuilder);
}
@@ -665,7 +665,7 @@ public abstract class AbstractCoordinator implements Closeable {
.setGenerationId(generation.generationId)
.setAssignments(groupAssignmentList)
);
- log.debug("Sending leader SyncGroup to coordinator {}: {}", this.coordinator, requestBuilder);
+ log.debug("Sending leader SyncGroup to coordinator {} at generation {}: {}", this.coordinator, this.generation, requestBuilder);
return sendSyncGroupRequest(requestBuilder);
} catch (RuntimeException e) {
return RequestFuture.failure(e);
@@ -819,36 +819,29 @@ public abstract class AbstractCoordinator implements Closeable {
}
/**
- * Get the current generation state if the group is stable.
- * @return the current generation or null if the group is unjoined/rebalancing
+ * Get the current generation state, regardless of whether it is currently stable.
+ * Note that the generation information can be updated while we are still in the middle
+ * of a rebalance, after the join-group response is received.
+ *
+ * @return the current generation
*/
protected synchronized Generation generation() {
- if (this.state != MemberState.STABLE)
- return null;
return generation;
}
- protected synchronized String memberId() {
- return generation == null ? JoinGroupRequest.UNKNOWN_MEMBER_ID :
- generation.memberId;
- }
-
/**
- * Check whether given generation id is matching the record within current generation.
- * Only using in unit tests.
- * @param generationId generation id
- * @return true if the two ids are matching.
+ * Get the current generation state if the group is stable, otherwise return null
+ *
+ * @return the current generation or null
*/
- final synchronized boolean hasMatchingGenerationId(int generationId) {
- return generation != null && generation.generationId == generationId;
+ protected synchronized Generation generationIfStable() {
+ if (this.state != MemberState.STABLE)
+ return null;
+ return generation;
}
- /**
- * @return true if the current generation's member ID is valid, false otherwise
- */
- // Visible for testing
- final synchronized boolean hasValidMemberId() {
- return generation != null && generation.hasMemberId();
+ protected synchronized String memberId() {
+ return generation.memberId;
}
private synchronized void resetGeneration() {
@@ -1348,12 +1341,35 @@ public abstract class AbstractCoordinator implements Closeable {
}
- // For testing only
+ // For testing only below
public Heartbeat heartbeat() {
return heartbeat;
}
- public void setLastRebalanceTime(final long timestamp) {
+ final void setLastRebalanceTime(final long timestamp) {
lastRebalanceEndMs = timestamp;
}
+
+ /**
+ * Check whether given generation id is matching the record within current generation.
+ *
+ * @param generationId generation id
+ * @return true if the two ids are matching.
+ */
+ final boolean hasMatchingGenerationId(int generationId) {
+ return generation != Generation.NO_GENERATION && generation.generationId == generationId;
+ }
+
+ final boolean hasUnknownGeneration() {
+ return generation == Generation.NO_GENERATION;
+ }
+
+ /**
+ * @return true if the current generation's member ID is valid, false otherwise
+ */
+ final boolean hasValidMemberId() {
+ return generation != Generation.NO_GENERATION && generation.hasMemberId();
+ }
+
+
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index 6b39acb..bceb9b8 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -582,7 +582,7 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
assignmentSnapshot = metadataSnapshot;
- log.debug("Finished assignment for group: {}", assignments);
+ log.info("Finished assignment for group at generation {}: {}", generation().generationId, assignments);
Map<String, ByteBuffer> groupAssignment = new HashMap<>();
for (Map.Entry<String, Assignment> assignmentEntry : assignments.entrySet()) {
@@ -764,8 +764,9 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
final Timer timer) {
if (partitions.isEmpty()) return Collections.emptyMap();
- final Generation generation = generation();
- if (pendingCommittedOffsetRequest != null && !pendingCommittedOffsetRequest.sameRequest(partitions, generation)) {
+ final Generation generationForOffsetRequest = generationIfStable();
+ if (pendingCommittedOffsetRequest != null &&
+ !pendingCommittedOffsetRequest.sameRequest(partitions, generationForOffsetRequest)) {
// if we were waiting for a different request, then just clear it.
pendingCommittedOffsetRequest = null;
}
@@ -779,7 +780,7 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
future = pendingCommittedOffsetRequest.response;
} else {
future = sendOffsetFetchRequest(partitions);
- pendingCommittedOffsetRequest = new PendingCommittedOffsetRequest(partitions, generation, future);
+ pendingCommittedOffsetRequest = new PendingCommittedOffsetRequest(partitions, generationForOffsetRequest, future);
}
client.poll(future, timer);
@@ -1039,7 +1040,7 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
final Generation generation;
if (subscriptions.partitionsAutoAssigned()) {
- generation = generation();
+ generation = generationIfStable();
// if the generation is null, we are not part of an active group (and we expect to be).
// the only thing we can do is fail the commit and let the user rejoin the group in poll()
if (generation == null) {
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
index 66786ca..d67fae9 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
@@ -413,7 +413,7 @@ public class AbstractCoordinatorTest {
assertTrue(consumerClient.poll(future, mockTime.timer(REQUEST_TIMEOUT_MS)));
assertEquals(Errors.UNKNOWN_MEMBER_ID.message(), future.exception().getMessage());
assertTrue(coordinator.rejoinNeededOrPending());
- assertTrue(coordinator.hasMatchingGenerationId(defaultGeneration));
+ assertTrue(coordinator.hasUnknownGeneration());
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
index 930eb54..fd3411b 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
@@ -987,7 +987,7 @@ public class ConsumerCoordinatorTest {
coordinator.maybeLeaveGroup("test maybe leave group");
assertTrue(received.get());
- AbstractCoordinator.Generation generation = coordinator.generation();
+ AbstractCoordinator.Generation generation = coordinator.generationIfStable();
assertNull(generation);
}
@@ -1865,7 +1865,7 @@ public class ConsumerCoordinatorTest {
AbstractCoordinator.Generation expectedGeneration = new AbstractCoordinator.Generation(1, consumerId, partitionAssignor.name());
assertFalse(coordinator.rejoinNeededOrPending());
- assertEquals(expectedGeneration, coordinator.generation());
+ assertEquals(expectedGeneration, coordinator.generationIfStable());
prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.REBALANCE_IN_PROGRESS);
@@ -1873,7 +1873,7 @@ public class ConsumerCoordinatorTest {
new OffsetAndMetadata(100L, "metadata")), time.timer(Long.MAX_VALUE)));
assertTrue(coordinator.rejoinNeededOrPending());
- assertEquals(expectedGeneration, coordinator.generation());
+ assertEquals(expectedGeneration, coordinator.generationIfStable());
}
@Test(expected = KafkaException.class)
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
index ff47e36..79940e0 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
@@ -221,7 +221,7 @@ public class WorkerCoordinator extends AbstractCoordinator implements Closeable
@Override
public String memberId() {
- Generation generation = generation();
+ Generation generation = generationIfStable();
if (generation != null)
return generation.memberId;
return JoinGroupRequest.UNKNOWN_MEMBER_ID;
[kafka] 01/02: KAFKA-9004;
Prevent older clients from fetching from a follower (#7531)
Posted by jg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch 2.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 0fbf012abda9e9b446c84dd66f605c7f5a0b8a8d
Author: David Arthur <mu...@gmail.com>
AuthorDate: Thu Oct 17 12:44:14 2019 -0400
KAFKA-9004; Prevent older clients from fetching from a follower (#7531)
With KIP-392, we allow consumers to fetch from followers. This capability is enabled when a replica selector has been provided in the configuration. When not in use, the intent is to preserve current behavior of fetching only from leader. The leader epoch is the mechanism that keeps us honest. When there is a leader change, the epoch gets bumped, consumer fetches fail due to the fenced epoch, and we find the new leader.
However, for old consumers, there is no similar protection. The leader epoch was not available to clients until recently. If there is a preferred leader election (for example), the old consumer will happily continue fetching from the demoted leader until a periodic metadata fetch causes us to discover the new leader. This does not create any problems from a correctness perspective–fetches are still bound by the high watermark–but it is unexpected and may cause unexpected performance c [...]
This patch fixes this problem by enforcing leader-only fetching for older versions of the fetch request.
Reviewers: Jason Gustafson <ja...@confluent.io>
---
.../main/scala/kafka/server/ReplicaManager.scala | 12 +--
.../unit/kafka/server/ReplicaManagerTest.scala | 94 +++++++++++++++++++++-
2 files changed, 98 insertions(+), 8 deletions(-)
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index d10c936..ff39339 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -848,18 +848,20 @@ class ReplicaManager(val config: KafkaConfig,
isolationLevel: IsolationLevel,
clientMetadata: Option[ClientMetadata]): Unit = {
val isFromFollower = Request.isValidBrokerId(replicaId)
-
- val fetchIsolation = if (isFromFollower || replicaId == Request.FutureLocalReplicaId)
+ val isFromConsumer = !(isFromFollower || replicaId == Request.FutureLocalReplicaId)
+ val fetchIsolation = if (!isFromConsumer)
FetchLogEnd
else if (isolationLevel == IsolationLevel.READ_COMMITTED)
FetchTxnCommitted
else
FetchHighWatermark
+ // Restrict fetching to leader if request is from follower or from a client with older version (no ClientMetadata)
+ val fetchOnlyFromLeader = isFromFollower || (isFromConsumer && clientMetadata.isEmpty)
def readFromLog(): Seq[(TopicPartition, LogReadResult)] = {
val result = readFromLocalLog(
replicaId = replicaId,
- fetchOnlyFromLeader = isFromFollower,
+ fetchOnlyFromLeader = fetchOnlyFromLeader,
fetchIsolation = fetchIsolation,
fetchMaxBytes = fetchMaxBytes,
hardMaxBytesLimit = hardMaxBytesLimit,
@@ -917,7 +919,7 @@ class ReplicaManager(val config: KafkaConfig,
fetchPartitionStatus += (topicPartition -> FetchPartitionStatus(logOffsetMetadata, partitionData))
})
}
- val fetchMetadata = FetchMetadata(fetchMinBytes, fetchMaxBytes, hardMaxBytesLimit, isFromFollower,
+ val fetchMetadata = FetchMetadata(fetchMinBytes, fetchMaxBytes, hardMaxBytesLimit, fetchOnlyFromLeader,
fetchIsolation, isFromFollower, replicaId, fetchPartitionStatus)
val delayedFetch = new DelayedFetch(timeout, fetchMetadata, this, quota, clientMetadata,
maybeUpdateHwAndSendResponse)
@@ -971,7 +973,7 @@ class ReplicaManager(val config: KafkaConfig,
s"${preferredReadReplica.get} for $clientMetadata")
}
// If a preferred read-replica is set, skip the read
- val offsetSnapshot = partition.fetchOffsetSnapshot(fetchInfo.currentLeaderEpoch, false)
+ val offsetSnapshot = partition.fetchOffsetSnapshot(fetchInfo.currentLeaderEpoch, fetchOnlyFromLeader = false)
LogReadResult(info = FetchDataInfo(LogOffsetMetadata.UnknownOffsetMetadata, MemoryRecords.EMPTY),
highWatermark = offsetSnapshot.highWatermark.messageOffset,
leaderLogStartOffset = offsetSnapshot.logStartOffset,
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index 484ee2d..ef05fa1 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -24,10 +24,8 @@ import java.util.concurrent.{CountDownLatch, TimeUnit}
import java.util.{Optional, Properties}
import kafka.api.Request
-import kafka.cluster.BrokerEndPoint
import kafka.log.{Log, LogConfig, LogManager, ProducerStateManager}
-import kafka.utils.{MockScheduler, MockTime, TestUtils}
-import TestUtils.createBroker
+import kafka.utils.TestUtils
import kafka.cluster.BrokerEndPoint
import kafka.server.QuotaFactory.UnboundedQuota
import kafka.server.checkpoints.LazyOffsetCheckpoints
@@ -921,6 +919,96 @@ class ReplicaManagerTest {
assertFalse(replicaManager.replicaSelectorOpt.isDefined)
}
+ @Test
+ def testOlderClientFetchFromLeaderOnly(): Unit = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer, aliveBrokerIds = Seq(0, 1))
+
+ val tp0 = new TopicPartition(topic, 0)
+ val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)
+ replicaManager.createPartition(tp0).createLogIfNotExists(0, isNew = false, isFutureReplica = false, offsetCheckpoints)
+ val partition0Replicas = Seq[Integer](0, 1).asJava
+ val leaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
+ Seq(
+ new LeaderAndIsrPartitionState()
+ .setTopicName(tp0.topic)
+ .setPartitionIndex(tp0.partition)
+ .setControllerEpoch(0)
+ .setLeader(1)
+ .setLeaderEpoch(0)
+ .setIsr(partition0Replicas)
+ .setZkVersion(0)
+ .setReplicas(partition0Replicas)
+ .setIsNew(true)).asJava,
+ Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
+ replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest, (_, _) => ())
+
+ def doFetch(replicaId: Int, partitionData: FetchRequest.PartitionData, clientMetadataOpt: Option[ClientMetadata]):
+ Option[FetchPartitionData] = {
+ var fetchResult: Option[FetchPartitionData] = None
+ def callback(response: Seq[(TopicPartition, FetchPartitionData)]): Unit = {
+ fetchResult = response.headOption.filter(_._1 == tp0).map(_._2)
+ }
+ replicaManager.fetchMessages(
+ timeout = 0L,
+ replicaId = replicaId,
+ fetchMinBytes = 1,
+ fetchMaxBytes = 100,
+ hardMaxBytesLimit = false,
+ fetchInfos = Seq(tp0 -> partitionData),
+ quota = UnboundedQuota,
+ isolationLevel = IsolationLevel.READ_UNCOMMITTED,
+ responseCallback = callback,
+ clientMetadata = clientMetadataOpt
+ )
+ fetchResult
+ }
+
+ // Fetch from follower, with non-empty ClientMetadata (FetchRequest v11+)
+ val clientMetadata = new DefaultClientMetadata("", "", null, KafkaPrincipal.ANONYMOUS, "")
+ var partitionData = new FetchRequest.PartitionData(0L, 0L, 100,
+ Optional.of(0))
+ var fetchResult = doFetch(Request.OrdinaryConsumerId, partitionData, Some(clientMetadata))
+ assertTrue(fetchResult.isDefined)
+ assertEquals(fetchResult.get.error, Errors.NONE)
+
+ // Fetch from follower, with empty ClientMetadata
+ fetchResult = None
+ partitionData = new FetchRequest.PartitionData(0L, 0L, 100,
+ Optional.of(0))
+ fetchResult = doFetch(Request.OrdinaryConsumerId, partitionData, None)
+ assertTrue(fetchResult.isDefined)
+ assertEquals(fetchResult.get.error, Errors.NOT_LEADER_FOR_PARTITION)
+
+ // Change to a leader, both cases are allowed
+ val leaderAndIsrRequest2 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
+ Seq(
+ new LeaderAndIsrPartitionState()
+ .setTopicName(tp0.topic)
+ .setPartitionIndex(tp0.partition)
+ .setControllerEpoch(0)
+ .setLeader(0)
+ .setLeaderEpoch(1)
+ .setIsr(partition0Replicas)
+ .setZkVersion(0)
+ .setReplicas(partition0Replicas)
+ .setIsNew(true)).asJava,
+ Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
+ replicaManager.becomeLeaderOrFollower(1, leaderAndIsrRequest2, (_, _) => ())
+
+ partitionData = new FetchRequest.PartitionData(0L, 0L, 100,
+ Optional.of(1))
+ fetchResult = doFetch(Request.OrdinaryConsumerId, partitionData, Some(clientMetadata))
+ assertTrue(fetchResult.isDefined)
+ assertEquals(fetchResult.get.error, Errors.NONE)
+
+ fetchResult = None
+ partitionData = new FetchRequest.PartitionData(0L, 0L, 100,
+ Optional.empty())
+ fetchResult = doFetch(Request.OrdinaryConsumerId, partitionData, None)
+ assertTrue(fetchResult.isDefined)
+ assertEquals(fetchResult.get.error, Errors.NONE)
+ }
+
/**
* This method assumes that the test using created ReplicaManager calls
* ReplicaManager.becomeLeaderOrFollower() once with LeaderAndIsrRequest containing