You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ij...@apache.org on 2017/06/06 02:37:53 UTC
kafka git commit: KAFKA-5376;
Ensure aborted transactions are propagated in DelayedFetch
Repository: kafka
Updated Branches:
refs/heads/trunk b119d0409 -> 2bb48a4fa
KAFKA-5376; Ensure aborted transactions are propagated in DelayedFetch
Author: Jason Gustafson <ja...@confluent.io>
Reviewers: Apurva Mehta <ap...@confluent.io>, Ismael Juma <is...@juma.me.uk>
Closes #3239 from hachikuji/KAFKA-5376
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/2bb48a4f
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/2bb48a4f
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/2bb48a4f
Branch: refs/heads/trunk
Commit: 2bb48a4fa45422d0773f2f4ea5b38328d3c173ac
Parents: b119d04
Author: Jason Gustafson <ja...@confluent.io>
Authored: Tue Jun 6 03:37:37 2017 +0100
Committer: Ismael Juma <is...@juma.me.uk>
Committed: Tue Jun 6 03:37:37 2017 +0100
----------------------------------------------------------------------
.../clients/consumer/internals/FetcherTest.java | 10 +-
core/src/main/scala/kafka/log/Log.scala | 11 +-
.../scala/kafka/log/ProducerStateManager.scala | 29 ++++
.../main/scala/kafka/server/DelayedFetch.scala | 3 +-
.../scala/kafka/server/ReplicaManager.scala | 2 +-
.../kafka/api/TransactionsTest.scala | 48 ++++++-
.../unit/kafka/server/ReplicaManagerTest.scala | 142 ++++++++++++++-----
7 files changed, 200 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/2bb48a4f/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
index 24eeb6f..0ed1139 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
@@ -41,6 +41,7 @@ import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
@@ -49,7 +50,6 @@ import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.record.ControlRecordType;
-import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.record.DefaultRecordBatch;
import org.apache.kafka.common.record.EndTransactionMarker;
import org.apache.kafka.common.record.LegacyRecord;
@@ -58,22 +58,22 @@ import org.apache.kafka.common.record.MemoryRecordsBuilder;
import org.apache.kafka.common.record.Record;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.record.SimpleRecord;
-import org.apache.kafka.common.requests.IsolationLevel;
-import org.apache.kafka.common.serialization.StringDeserializer;
-import org.apache.kafka.common.utils.ByteBufferOutputStream;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.requests.FetchRequest;
+import org.apache.kafka.common.requests.FetchRequest.PartitionData;
import org.apache.kafka.common.requests.FetchResponse;
+import org.apache.kafka.common.requests.IsolationLevel;
import org.apache.kafka.common.requests.ListOffsetRequest;
import org.apache.kafka.common.requests.ListOffsetResponse;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.requests.ResponseHeader;
-import org.apache.kafka.common.requests.FetchRequest.PartitionData;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.test.DelayedReceive;
http://git-wip-us.apache.org/repos/asf/kafka/blob/2bb48a4f/core/src/main/scala/kafka/log/Log.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala
index 5522508..4bd2c2c 100644
--- a/core/src/main/scala/kafka/log/Log.scala
+++ b/core/src/main/scala/kafka/log/Log.scala
@@ -89,7 +89,15 @@ case class LogAppendInfo(var firstOffset: Long,
* COMMIT/ABORT control record which indicates the transaction's completion.
* @param isAborted Whether or not the transaction was aborted
*/
-case class CompletedTxn(producerId: Long, firstOffset: Long, lastOffset: Long, isAborted: Boolean)
+case class CompletedTxn(producerId: Long, firstOffset: Long, lastOffset: Long, isAborted: Boolean) {
+ override def toString: String = {
+ "CompletedTxn(" +
+ s"producerId=$producerId, " +
+ s"firstOffset=$firstOffset, " +
+ s"lastOffset=$lastOffset, " +
+ s"isAborted=$isAborted)"
+ }
+}
/**
* An append-only log for storing messages.
@@ -631,7 +639,6 @@ class Log(@volatile var dir: File,
// update the producer state
for ((producerId, producerAppendInfo) <- updatedProducers) {
- trace(s"Updating producer $producerId state: ${producerAppendInfo.lastEntry}")
producerAppendInfo.maybeCacheTxnFirstOffsetMetadata(logOffsetMetadata)
producerStateManager.update(producerAppendInfo)
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/2bb48a4f/core/src/main/scala/kafka/log/ProducerStateManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/ProducerStateManager.scala b/core/src/main/scala/kafka/log/ProducerStateManager.scala
index d6e704a..e791052 100644
--- a/core/src/main/scala/kafka/log/ProducerStateManager.scala
+++ b/core/src/main/scala/kafka/log/ProducerStateManager.scala
@@ -38,6 +38,13 @@ class CorruptSnapshotException(msg: String) extends KafkaException(msg)
private[log] case class TxnMetadata(producerId: Long, var firstOffset: LogOffsetMetadata, var lastOffset: Option[Long] = None) {
def this(producerId: Long, firstOffset: Long) = this(producerId, LogOffsetMetadata(firstOffset))
+
+ override def toString: String = {
+ "TxnMetadata(" +
+ s"producerId=$producerId, " +
+ s"firstOffset=$firstOffset, " +
+ s"lastOffset=$lastOffset)"
+ }
}
private[log] object ProducerIdEntry {
@@ -56,6 +63,16 @@ private[log] case class ProducerIdEntry(producerId: Long, producerEpoch: Short,
batch.baseSequence == firstSeq &&
batch.lastSequence == lastSeq
}
+
+ override def toString: String = {
+ "ProducerIdEntry(" +
+ s"producerId=$producerId, " +
+ s"producerEpoch=$producerEpoch, " +
+ s"firstSequence=$firstSeq, " +
+ s"lastSequence=$lastSeq, " +
+ s"currentTxnFirstOffset=$currentTxnFirstOffset, " +
+ s"coordinatorEpoch=$coordinatorEpoch)"
+ }
}
/**
@@ -220,6 +237,16 @@ private[log] class ProducerAppendInfo(val producerId: Long,
}
}
+ override def toString: String = {
+ "ProducerAppendInfo(" +
+ s"producerId=$producerId, " +
+ s"producerEpoch=$producerEpoch, " +
+ s"firstSequence=$firstSeq, " +
+ s"lastSequence=$lastSeq, " +
+ s"currentTxnFirstOffset=$currentTxnFirstOffset, " +
+ s"coordinatorEpoch=$coordinatorEpoch, " +
+ s"startedTransactions=$transactions)"
+ }
}
object ProducerStateManager {
@@ -481,6 +508,8 @@ class ProducerStateManager(val topicPartition: TopicPartition,
if (appendInfo.producerId == RecordBatch.NO_PRODUCER_ID)
throw new IllegalArgumentException(s"Invalid producer id ${appendInfo.producerId} passed to update")
+ trace(s"Updated producer ${appendInfo.producerId} state to $appendInfo")
+
val entry = appendInfo.lastEntry
producers.put(appendInfo.producerId, entry)
appendInfo.startedTransactions.foreach { txn =>
http://git-wip-us.apache.org/repos/asf/kafka/blob/2bb48a4f/core/src/main/scala/kafka/server/DelayedFetch.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/DelayedFetch.scala b/core/src/main/scala/kafka/server/DelayedFetch.scala
index c5cdf57..13a01bc 100644
--- a/core/src/main/scala/kafka/server/DelayedFetch.scala
+++ b/core/src/main/scala/kafka/server/DelayedFetch.scala
@@ -156,7 +156,8 @@ class DelayedFetch(delayMs: Long,
isolationLevel = isolationLevel)
val fetchPartitionData = logReadResults.map { case (tp, result) =>
- tp -> FetchPartitionData(result.error, result.hw, result.leaderLogStartOffset, result.info.records)
+ tp -> FetchPartitionData(result.error, result.hw, result.leaderLogStartOffset, result.info.records,
+ result.info.abortedTransactions)
}
responseCallback(fetchPartitionData)
http://git-wip-us.apache.org/repos/asf/kafka/blob/2bb48a4f/core/src/main/scala/kafka/server/ReplicaManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index a8d3e94..d2670b7 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -94,7 +94,7 @@ case class LogReadResult(info: FetchDataInfo,
}
case class FetchPartitionData(error: Errors = Errors.NONE, hw: Long = -1L, logStartOffset: Long, records: Records,
- abortedTransactions: Option[List[AbortedTransaction]] = None)
+ abortedTransactions: Option[List[AbortedTransaction]])
object LogReadResult {
val UnknownLogReadResult = LogReadResult(info = FetchDataInfo(LogOffsetMetadata.UnknownOffsetMetadata, MemoryRecords.EMPTY),
http://git-wip-us.apache.org/repos/asf/kafka/blob/2bb48a4f/core/src/test/scala/integration/kafka/api/TransactionsTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/TransactionsTest.scala b/core/src/test/scala/integration/kafka/api/TransactionsTest.scala
index 4593d9c..8762ea8 100644
--- a/core/src/test/scala/integration/kafka/api/TransactionsTest.scala
+++ b/core/src/test/scala/integration/kafka/api/TransactionsTest.scala
@@ -160,6 +160,50 @@ class TransactionsTest extends KafkaServerTestHarness {
}
@Test
+ def testDelayedFetchIncludesAbortedTransaction(): Unit = {
+ val producer1 = transactionalProducers.head
+ val producer2 = createTransactionalProducer("other")
+
+ producer1.initTransactions()
+ producer2.initTransactions()
+
+ producer1.beginTransaction()
+ producer2.beginTransaction()
+ producer2.send(new ProducerRecord(topic1, 0, "x".getBytes, "1".getBytes))
+ producer2.flush()
+
+ producer1.send(new ProducerRecord(topic1, 0, "y".getBytes, "1".getBytes))
+ producer1.send(new ProducerRecord(topic1, 0, "y".getBytes, "2".getBytes))
+ producer1.flush()
+
+ producer2.send(new ProducerRecord(topic1, 0, "x".getBytes, "2".getBytes))
+ producer2.flush()
+
+ producer1.abortTransaction()
+ producer2.commitTransaction()
+
+ // ensure that the consumer's fetch will sit in purgatory
+ val consumerProps = new Properties()
+ consumerProps.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "100000")
+ consumerProps.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "100")
+ val readCommittedConsumer = createReadCommittedConsumer(props = consumerProps)
+
+ readCommittedConsumer.assign(Set(new TopicPartition(topic1, 0)).asJava)
+ val records = consumeRecords(readCommittedConsumer, numMessages = 2)
+ assertEquals(2, records.size)
+
+ val first = records.head
+ assertEquals("x", new String(first.key))
+ assertEquals("1", new String(first.value))
+ assertEquals(0L, first.offset)
+
+ val second = records.last
+ assertEquals("x", new String(second.key))
+ assertEquals("2", new String(second.value))
+ assertEquals(3L, second.offset)
+ }
+
+ @Test
def testSendOffsets() = {
// The basic plan for the test is as follows:
// 1. Seed topic1 with 1000 unique, numbered, messages.
@@ -406,8 +450,8 @@ class TransactionsTest extends KafkaServerTestHarness {
serverProps
}
- private def createReadCommittedConsumer(group: String = "group", maxPollRecords: Int = 500) = {
- val props = new Properties()
+ private def createReadCommittedConsumer(group: String = "group", maxPollRecords: Int = 500,
+ props: Properties = new Properties) = {
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed")
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords.toString)
http://git-wip-us.apache.org/repos/asf/kafka/blob/2bb48a4f/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index a33968a..45f62d7 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -129,7 +129,6 @@ class ReplicaManagerTest {
props.put("log.dir", TestUtils.tempRelativeDir("data").getAbsolutePath)
val config = KafkaConfig.fromProps(props)
val logProps = new Properties()
- logProps.put(LogConfig.MessageTimestampDifferenceMaxMsProp, Long.MaxValue.toString)
val mockLogMgr = TestUtils.createLogManager(config.logDirs.map(new File(_)).toArray, LogConfig(logProps))
val aliveBrokers = Seq(createBroker(0, "host0", 0), createBroker(1, "host1", 1))
val metadataCache = EasyMock.createMock(classOf[MetadataCache])
@@ -154,8 +153,8 @@ class ReplicaManagerTest {
fetchCallbackFired = true
}
- val brokerList: java.util.List[Integer] = Seq[Integer](0, 1).asJava
- val brokerSet: java.util.Set[Integer] = Set[Integer](0, 1).asJava
+ val brokerList = Seq[Integer](0, 1).asJava
+ val brokerSet = Set[Integer](0, 1).asJava
val partition = rm.getOrCreatePartition(new TopicPartition(topic, 0))
partition.getOrCreateReplica(0)
@@ -163,7 +162,7 @@ class ReplicaManagerTest {
val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(0, 0,
collection.immutable.Map(new TopicPartition(topic, 0) -> new PartitionState(0, 0, 0, brokerList, 0, brokerSet)).asJava,
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
- rm.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => {})
+ rm.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ())
rm.getLeaderReplicaIfLocal(new TopicPartition(topic, 0))
val records = MemoryRecords.withRecords(CompressionType.NONE, new SimpleRecord("first message".getBytes()))
@@ -177,7 +176,7 @@ class ReplicaManagerTest {
val leaderAndIsrRequest2 = new LeaderAndIsrRequest.Builder(0, 0,
collection.immutable.Map(new TopicPartition(topic, 0) -> new PartitionState(0, 1, 1, brokerList, 0, brokerSet)).asJava,
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
- rm.becomeLeaderOrFollower(1, leaderAndIsrRequest2, (_, _) => {})
+ rm.becomeLeaderOrFollower(1, leaderAndIsrRequest2, (_, _) => ())
assertTrue(produceCallbackFired)
assertTrue(fetchCallbackFired)
@@ -188,35 +187,12 @@ class ReplicaManagerTest {
@Test
def testReadCommittedFetchLimitedAtLSO(): Unit = {
- val props = TestUtils.createBrokerConfig(1, TestUtils.MockZkConnect)
- props.put("log.dir", TestUtils.tempRelativeDir("data").getAbsolutePath)
- props.put("broker.id", Int.box(0))
- val config = KafkaConfig.fromProps(props)
- val logProps = new Properties()
- logProps.put(LogConfig.MessageTimestampDifferenceMaxMsProp, Long.MaxValue.toString)
- val mockLogMgr = TestUtils.createLogManager(config.logDirs.map(new File(_)).toArray, LogConfig(logProps))
- val aliveBrokers = Seq(createBroker(0, "host0", 0), createBroker(1, "host1", 1))
- val metadataCache = EasyMock.createMock(classOf[MetadataCache])
- EasyMock.expect(metadataCache.getAliveBrokers).andReturn(aliveBrokers).anyTimes()
- EasyMock.expect(metadataCache.isBrokerAlive(EasyMock.eq(0))).andReturn(true).anyTimes()
- EasyMock.expect(metadataCache.isBrokerAlive(EasyMock.eq(1))).andReturn(true).anyTimes()
- EasyMock.replay(metadataCache)
-
val timer = new MockTimer
- val mockProducePurgatory = new DelayedOperationPurgatory[DelayedProduce](
- purgatoryName = "Produce", timer, reaperEnabled = false)
- val mockFetchPurgatory = new DelayedOperationPurgatory[DelayedFetch](
- purgatoryName = "Fetch", timer, reaperEnabled = false)
- val mockDeleteRecordsPurgatory = new DelayedOperationPurgatory[DelayedDeleteRecords](
- purgatoryName = "DeleteRecords", timer, reaperEnabled = false)
-
- val replicaManager = new ReplicaManager(config, metrics, time, zkUtils, new MockScheduler(time), mockLogMgr,
- new AtomicBoolean(false), QuotaFactory.instantiate(config, metrics, time).follower, new BrokerTopicStats,
- metadataCache, mockProducePurgatory, mockFetchPurgatory, mockDeleteRecordsPurgatory, Option(this.getClass.getName))
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(timer)
try {
- val brokerList: java.util.List[Integer] = Seq[Integer](0, 1).asJava
- val brokerSet: java.util.Set[Integer] = Set[Integer](0, 1).asJava
+ val brokerList = Seq[Integer](0, 1).asJava
+ val brokerSet = Set[Integer](0, 1).asJava
val partition = replicaManager.getOrCreatePartition(new TopicPartition(topic, 0))
partition.getOrCreateReplica(0)
@@ -225,7 +201,7 @@ class ReplicaManagerTest {
val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(0, 0,
collection.immutable.Map(new TopicPartition(topic, 0) -> new PartitionState(0, 0, 0, brokerList, 0, brokerSet)).asJava,
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
- replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => {})
+ replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ())
replicaManager.getLeaderReplicaIfLocal(new TopicPartition(topic, 0))
def produceCallback(responseStatus: Map[TopicPartition, PartitionResponse]) =
@@ -308,13 +284,85 @@ class ReplicaManagerTest {
}
@Test
+ def testDelayedFetchIncludesAbortedTransactions(): Unit = {
+ val timer = new MockTimer
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(timer)
+
+ try {
+ val brokerList: java.util.List[Integer] = Seq[Integer](0, 1).asJava
+ val brokerSet: java.util.Set[Integer] = Set[Integer](0, 1).asJava
+
+ val partition = replicaManager.getOrCreatePartition(new TopicPartition(topic, 0))
+ partition.getOrCreateReplica(0)
+
+ // Make this replica the leader.
+ val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(0, 0,
+ collection.immutable.Map(new TopicPartition(topic, 0) -> new PartitionState(0, 0, 0, brokerList, 0, brokerSet)).asJava,
+ Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
+ replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ())
+ replicaManager.getLeaderReplicaIfLocal(new TopicPartition(topic, 0))
+
+ def produceCallback(responseStatus: Map[TopicPartition, PartitionResponse]) =
+ responseStatus.values.foreach { status =>
+ assertEquals(Errors.NONE, status.error)
+ }
+
+ val producerId = 234L
+ val epoch = 5.toShort
+
+ // write a few batches as part of a transaction
+ val numRecords = 3
+ for (sequence <- 0 until numRecords) {
+ val records = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId, epoch, sequence,
+ new SimpleRecord(s"message $sequence".getBytes))
+ appendRecords(replicaManager, Map(new TopicPartition(topic, 0) -> records), produceCallback)
+ }
+
+ // now abort the transaction
+ val endTxnMarker = new EndTransactionMarker(ControlRecordType.ABORT, 0)
+ val abortRecordBatch = MemoryRecords.withEndTransactionMarker(producerId, epoch, endTxnMarker)
+ appendRecords(replicaManager, Map(new TopicPartition(topic, 0) -> abortRecordBatch), produceCallback,
+ isFromClient = false)
+
+ // fetch as follower to advance the high watermark
+ fetchAsFollower(replicaManager, Seq(new TopicPartition(topic, 0) -> new PartitionData(numRecords + 1, 0, 100000)),
+ responseState => (), isolationLevel = IsolationLevel.READ_UNCOMMITTED)
+
+ var fetchDataOpt: Option[FetchPartitionData] = None
+ def fetchCallback(responseStatus: Seq[(TopicPartition, FetchPartitionData)]) = {
+ fetchDataOpt = Some(responseStatus.map(_._2).head)
+ }
+
+ // Set the minBytes in order force this request to enter purgatory. When it returns, we should still
+ // see the newly aborted transaction.
+ fetchAsConsumer(replicaManager, Seq(new TopicPartition(topic, 0) -> new PartitionData(0, 0, 100000)),
+ fetchCallback, isolationLevel = IsolationLevel.READ_COMMITTED, minBytes = 10000)
+ assertTrue(fetchDataOpt.isEmpty)
+
+ timer.advanceClock(1001)
+ assertTrue(fetchDataOpt.isDefined)
+
+ val fetchData = fetchDataOpt.get
+ assertEquals(Errors.NONE, fetchData.error)
+ assertEquals(numRecords + 1, fetchData.records.records.asScala.size)
+ assertTrue(fetchData.abortedTransactions.isDefined)
+ assertEquals(1, fetchData.abortedTransactions.get.size)
+
+ val abortedTransaction = fetchData.abortedTransactions.get.head
+ assertEquals(0L, abortedTransaction.firstOffset)
+ assertEquals(producerId, abortedTransaction.producerId)
+ } finally {
+ replicaManager.shutdown(checkpointHW = false)
+ }
+ }
+
+ @Test
def testFetchBeyondHighWatermarkReturnEmptyResponse() {
val props = TestUtils.createBrokerConfig(1, TestUtils.MockZkConnect)
props.put("log.dir", TestUtils.tempRelativeDir("data").getAbsolutePath)
props.put("broker.id", Int.box(0))
val config = KafkaConfig.fromProps(props)
val logProps = new Properties()
- logProps.put(LogConfig.MessageTimestampDifferenceMaxMsProp, Long.MaxValue.toString)
val mockLogMgr = TestUtils.createLogManager(config.logDirs.map(new File(_)).toArray, LogConfig(logProps))
val aliveBrokers = Seq(createBroker(0, "host0", 0), createBroker(1, "host1", 1), createBroker(1, "host2", 2))
val metadataCache = EasyMock.createMock(classOf[MetadataCache])
@@ -338,7 +386,7 @@ class ReplicaManagerTest {
val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(0, 0,
collection.immutable.Map(new TopicPartition(topic, 0) -> new PartitionState(0, 0, 0, brokerList, 0, brokerSet)).asJava,
Set(new Node(0, "host1", 0), new Node(1, "host2", 1), new Node(2, "host2", 2)).asJava).build()
- rm.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => {})
+ rm.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ())
rm.getLeaderReplicaIfLocal(new TopicPartition(topic, 0))
def produceCallback(responseStatus: Map[TopicPartition, PartitionResponse]) = {}
@@ -421,4 +469,30 @@ class ReplicaManagerTest {
isolationLevel = isolationLevel)
}
+ private def setupReplicaManagerWithMockedPurgatories(timer: MockTimer): ReplicaManager = {
+ val props = TestUtils.createBrokerConfig(1, TestUtils.MockZkConnect)
+ props.put("log.dir", TestUtils.tempRelativeDir("data").getAbsolutePath)
+ props.put("broker.id", Int.box(0))
+ val config = KafkaConfig.fromProps(props)
+ val logProps = new Properties()
+ val mockLogMgr = TestUtils.createLogManager(config.logDirs.map(new File(_)).toArray, LogConfig(logProps))
+ val aliveBrokers = Seq(createBroker(0, "host0", 0), createBroker(1, "host1", 1))
+ val metadataCache = EasyMock.createMock(classOf[MetadataCache])
+ EasyMock.expect(metadataCache.getAliveBrokers).andReturn(aliveBrokers).anyTimes()
+ EasyMock.expect(metadataCache.isBrokerAlive(EasyMock.eq(0))).andReturn(true).anyTimes()
+ EasyMock.expect(metadataCache.isBrokerAlive(EasyMock.eq(1))).andReturn(true).anyTimes()
+ EasyMock.replay(metadataCache)
+
+ val mockProducePurgatory = new DelayedOperationPurgatory[DelayedProduce](
+ purgatoryName = "Produce", timer, reaperEnabled = false)
+ val mockFetchPurgatory = new DelayedOperationPurgatory[DelayedFetch](
+ purgatoryName = "Fetch", timer, reaperEnabled = false)
+ val mockDeleteRecordsPurgatory = new DelayedOperationPurgatory[DelayedDeleteRecords](
+ purgatoryName = "DeleteRecords", timer, reaperEnabled = false)
+
+ new ReplicaManager(config, metrics, time, zkUtils, new MockScheduler(time), mockLogMgr,
+ new AtomicBoolean(false), QuotaFactory.instantiate(config, metrics, time).follower, new BrokerTopicStats,
+ metadataCache, mockProducePurgatory, mockFetchPurgatory, mockDeleteRecordsPurgatory, Option(this.getClass.getName))
+ }
+
}