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 2017/01/04 03:53:40 UTC
[1/2] kafka git commit: MINOR: Improvements to Record related classes
(part 1)
Repository: kafka
Updated Branches:
refs/heads/trunk ce1cb329d -> 6d6c77a7a
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/log/LogSegment.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/LogSegment.scala b/core/src/main/scala/kafka/log/LogSegment.scala
index eddb47a..8854c3a 100755
--- a/core/src/main/scala/kafka/log/LogSegment.scala
+++ b/core/src/main/scala/kafka/log/LogSegment.scala
@@ -312,7 +312,7 @@ class LogSegment(val log: FileRecords,
if (ms == null) {
baseOffset
} else {
- ms.records.shallowEntries.asScala.toSeq.lastOption match {
+ ms.records.shallowEntries.asScala.lastOption match {
case None => baseOffset
case Some(last) => last.nextOffset
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/log/LogValidator.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/LogValidator.scala b/core/src/main/scala/kafka/log/LogValidator.scala
index d99c2ad..224a792 100644
--- a/core/src/main/scala/kafka/log/LogValidator.scala
+++ b/core/src/main/scala/kafka/log/LogValidator.scala
@@ -64,76 +64,8 @@ private[kafka] object LogValidator {
assignOffsetsNonCompressed(records, offsetCounter, now, compactedTopic, messageTimestampType,
messageTimestampDiffMaxMs)
} else {
- // Deal with compressed messages
- // We cannot do in place assignment in one of the following situations:
- // 1. Source and target compression codec are different
- // 2. When magic value to use is 0 because offsets need to be overwritten
- // 3. When magic value to use is above 0, but some fields of inner messages need to be overwritten.
- // 4. Message format conversion is needed.
-
- // No in place assignment situation 1 and 2
- var inPlaceAssignment = sourceCodec == targetCodec && messageFormatVersion > Record.MAGIC_VALUE_V0
-
- var maxTimestamp = Record.NO_TIMESTAMP
- val expectedInnerOffset = new LongRef(0)
- val validatedRecords = new mutable.ArrayBuffer[Record]
-
- records.deepEntries(true).asScala.foreach { logEntry =>
- val record = logEntry.record
- validateKey(record, compactedTopic)
-
- if (record.magic > Record.MAGIC_VALUE_V0 && messageFormatVersion > Record.MAGIC_VALUE_V0) {
- // No in place assignment situation 3
- // Validate the timestamp
- validateTimestamp(record, now, messageTimestampType, messageTimestampDiffMaxMs)
- // Check if we need to overwrite offset
- if (logEntry.offset != expectedInnerOffset.getAndIncrement())
- inPlaceAssignment = false
- if (record.timestamp > maxTimestamp)
- maxTimestamp = record.timestamp
- }
-
- if (sourceCodec != NoCompressionCodec && logEntry.isCompressed)
- throw new InvalidMessageException("Compressed outer record should not have an inner record with a " +
- s"compression attribute set: $record")
-
- // No in place assignment situation 4
- if (record.magic != messageFormatVersion)
- inPlaceAssignment = false
-
- validatedRecords += record.convert(messageFormatVersion)
- }
-
- if (!inPlaceAssignment) {
- val entries = validatedRecords.map(record => LogEntry.create(offsetCounter.getAndIncrement(), record))
- val builder = MemoryRecords.builderWithEntries(messageTimestampType, CompressionType.forId(targetCodec.codec),
- now, entries.asJava)
- builder.close()
- val info = builder.info
-
- ValidationAndOffsetAssignResult(
- validatedRecords = builder.build(),
- maxTimestamp = info.maxTimestamp,
- shallowOffsetOfMaxTimestamp = info.shallowOffsetOfMaxTimestamp,
- messageSizeMaybeChanged = true)
- } else {
- // ensure the inner messages are valid
- validatedRecords.foreach(_.ensureValid)
-
- // we can update the wrapper message only and write the compressed payload as is
- val entry = records.shallowEntries.iterator.next()
- val offset = offsetCounter.addAndGet(validatedRecords.size) - 1
- entry.setOffset(offset)
- if (messageTimestampType == TimestampType.CREATE_TIME)
- entry.setCreateTime(maxTimestamp)
- else if (messageTimestampType == TimestampType.LOG_APPEND_TIME)
- entry.setLogAppendTime(now)
-
- ValidationAndOffsetAssignResult(validatedRecords = records,
- maxTimestamp = if (messageTimestampType == TimestampType.LOG_APPEND_TIME) now else maxTimestamp,
- shallowOffsetOfMaxTimestamp = offset,
- messageSizeMaybeChanged = false)
- }
+ validateMessagesAndAssignOffsetsCompressed(records, offsetCounter, now, sourceCodec, targetCodec, compactedTopic,
+ messageFormatVersion, messageTimestampType, messageTimestampDiffMaxMs)
}
}
@@ -159,11 +91,10 @@ private[kafka] object LogValidator {
builder.convertAndAppendWithOffset(offsetCounter.getAndIncrement(), record)
}
- builder.close()
+ val convertedRecords = builder.build()
val info = builder.info
-
ValidationAndOffsetAssignResult(
- validatedRecords = builder.build(),
+ validatedRecords = convertedRecords,
maxTimestamp = info.maxTimestamp,
shallowOffsetOfMaxTimestamp = info.shallowOffsetOfMaxTimestamp,
messageSizeMaybeChanged = true)
@@ -210,6 +141,87 @@ private[kafka] object LogValidator {
messageSizeMaybeChanged = false)
}
+ /**
+ * We cannot do in place assignment in one of the following situations:
+ * 1. Source and target compression codec are different
+ * 2. When magic value to use is 0 because offsets need to be overwritten
+ * 3. When magic value to use is above 0, but some fields of inner messages need to be overwritten.
+ * 4. Message format conversion is needed.
+ */
+ def validateMessagesAndAssignOffsetsCompressed(records: MemoryRecords,
+ offsetCounter: LongRef,
+ now: Long,
+ sourceCodec: CompressionCodec,
+ targetCodec: CompressionCodec,
+ compactedTopic: Boolean = false,
+ messageFormatVersion: Byte = Record.CURRENT_MAGIC_VALUE,
+ messageTimestampType: TimestampType,
+ messageTimestampDiffMaxMs: Long): ValidationAndOffsetAssignResult = {
+ // No in place assignment situation 1 and 2
+ var inPlaceAssignment = sourceCodec == targetCodec && messageFormatVersion > Record.MAGIC_VALUE_V0
+
+ var maxTimestamp = Record.NO_TIMESTAMP
+ val expectedInnerOffset = new LongRef(0)
+ val validatedRecords = new mutable.ArrayBuffer[Record]
+
+ records.deepEntries(true).asScala.foreach { logEntry =>
+ val record = logEntry.record
+ validateKey(record, compactedTopic)
+
+ if (record.magic > Record.MAGIC_VALUE_V0 && messageFormatVersion > Record.MAGIC_VALUE_V0) {
+ // Validate the timestamp
+ validateTimestamp(record, now, messageTimestampType, messageTimestampDiffMaxMs)
+ // Check if we need to overwrite offset, no in place assignment situation 3
+ if (logEntry.offset != expectedInnerOffset.getAndIncrement())
+ inPlaceAssignment = false
+ if (record.timestamp > maxTimestamp)
+ maxTimestamp = record.timestamp
+ }
+
+ if (sourceCodec != NoCompressionCodec && logEntry.isCompressed)
+ throw new InvalidMessageException("Compressed outer record should not have an inner record with a " +
+ s"compression attribute set: $record")
+
+ // No in place assignment situation 4
+ if (record.magic != messageFormatVersion)
+ inPlaceAssignment = false
+
+ validatedRecords += record.convert(messageFormatVersion)
+ }
+
+ if (!inPlaceAssignment) {
+ val entries = validatedRecords.map(record => LogEntry.create(offsetCounter.getAndIncrement(), record))
+ val builder = MemoryRecords.builderWithEntries(messageTimestampType, CompressionType.forId(targetCodec.codec),
+ now, entries.asJava)
+ val updatedRecords = builder.build()
+ val info = builder.info
+ ValidationAndOffsetAssignResult(
+ validatedRecords = updatedRecords,
+ maxTimestamp = info.maxTimestamp,
+ shallowOffsetOfMaxTimestamp = info.shallowOffsetOfMaxTimestamp,
+ messageSizeMaybeChanged = true)
+ } else {
+ // ensure the inner messages are valid
+ validatedRecords.foreach(_.ensureValid)
+
+ // we can update the wrapper message only and write the compressed payload as is
+ val entry = records.shallowEntries.iterator.next()
+ val offset = offsetCounter.addAndGet(validatedRecords.size) - 1
+ entry.setOffset(offset)
+
+ val shallowTimestamp = if (messageTimestampType == TimestampType.LOG_APPEND_TIME) now else maxTimestamp
+ if (messageTimestampType == TimestampType.LOG_APPEND_TIME)
+ entry.setLogAppendTime(shallowTimestamp)
+ else if (messageTimestampType == TimestampType.CREATE_TIME)
+ entry.setCreateTime(shallowTimestamp)
+
+ ValidationAndOffsetAssignResult(validatedRecords = records,
+ maxTimestamp = shallowTimestamp,
+ shallowOffsetOfMaxTimestamp = offset,
+ messageSizeMaybeChanged = false)
+ }
+ }
+
private def validateKey(record: Record, compactedTopic: Boolean) {
if (compactedTopic && !record.hasKey)
throw new InvalidMessageException("Compacted topic cannot accept message without key.")
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/message/Message.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/message/Message.scala b/core/src/main/scala/kafka/message/Message.scala
index 175b7e9..e0efb3d 100755
--- a/core/src/main/scala/kafka/message/Message.scala
+++ b/core/src/main/scala/kafka/message/Message.scala
@@ -99,8 +99,8 @@ object Message {
def fromRecord(record: Record): Message = {
- val wrapperTimestamp: Option[Long] = if (record.wrapperRecordTimestamp() == null) None else Some(record.wrapperRecordTimestamp())
- val wrapperTimestampType = Option(record.wrapperRecordTimestampType())
+ val wrapperTimestamp: Option[Long] = if (record.wrapperRecordTimestamp == null) None else Some(record.wrapperRecordTimestamp)
+ val wrapperTimestampType = Option(record.wrapperRecordTimestampType)
new Message(record.buffer, wrapperTimestamp, wrapperTimestampType)
}
}
@@ -139,13 +139,9 @@ class Message(val buffer: ByteBuffer,
import kafka.message.Message._
- private[message] def asRecord: Record = {
- wrapperMessageTimestamp match {
- case None => new Record(buffer)
- case Some(timestamp) =>
- val timestampType = wrapperMessageTimestampType.orNull
- new Record(buffer, timestamp, timestampType)
- }
+ private[message] def asRecord: Record = wrapperMessageTimestamp match {
+ case None => new Record(buffer)
+ case Some(timestamp) => new Record(buffer, timestamp, wrapperMessageTimestampType.orNull)
}
/**
@@ -227,7 +223,7 @@ class Message(val buffer: ByteBuffer,
* Compute the checksum of the message from the message contents
*/
def computeChecksum: Long =
- CoreUtils.crc32(buffer.array, buffer.arrayOffset + MagicOffset, buffer.limit - MagicOffset)
+ Utils.computeChecksum(buffer, MagicOffset, buffer.limit - MagicOffset)
/**
* Retrieve the previously computed CRC for this message
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
index 1dbd373..ec25700 100755
--- a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
+++ b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
@@ -145,7 +145,7 @@ abstract class AbstractFetcherThread(name: String,
case Errors.NONE =>
try {
val records = partitionData.toRecords
- val newOffset = records.shallowEntries.asScala.toSeq.lastOption.map(_.nextOffset).getOrElse(
+ val newOffset = records.shallowEntries.asScala.lastOption.map(_.nextOffset).getOrElse(
currentPartitionFetchState.offset)
fetcherLagStats.getAndMaybePut(topic, partitionId).lag = Math.max(0L, partitionData.highWatermark - newOffset)
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/server/DelayedOperationKey.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/DelayedOperationKey.scala b/core/src/main/scala/kafka/server/DelayedOperationKey.scala
index 0e05cce..1933339 100644
--- a/core/src/main/scala/kafka/server/DelayedOperationKey.scala
+++ b/core/src/main/scala/kafka/server/DelayedOperationKey.scala
@@ -36,8 +36,6 @@ case class TopicPartitionOperationKey(topic: String, partition: Int) extends Del
def this(topicPartition: TopicPartition) = this(topicPartition.topic, topicPartition.partition)
- def this(topicAndPartition: TopicAndPartition) = this(topicAndPartition.topic, topicAndPartition.partition)
-
override def keyLabel = "%s-%d".format(topic, partition)
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/server/KafkaServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index bbddfae..d78021f 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -427,7 +427,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
val controlledShutdownRequest = new ControlledShutdownRequest(config.brokerId)
val request = new ClientRequest(node(prevController).idString, time.milliseconds(), true,
requestHeader, controlledShutdownRequest, null)
- val clientResponse = networkClient.blockingSendAndReceive(request, controlledShutdownRequest)(time)
+ val clientResponse = networkClient.blockingSendAndReceive(request)(time)
val shutdownResponse = clientResponse.responseBody.asInstanceOf[ControlledShutdownResponse]
if (shutdownResponse.errorCode == Errors.NONE.code && shutdownResponse.partitionsRemaining.isEmpty) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/server/LogOffsetMetadata.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/LogOffsetMetadata.scala b/core/src/main/scala/kafka/server/LogOffsetMetadata.scala
index 7067b20..05e9842 100644
--- a/core/src/main/scala/kafka/server/LogOffsetMetadata.scala
+++ b/core/src/main/scala/kafka/server/LogOffsetMetadata.scala
@@ -25,7 +25,7 @@ object LogOffsetMetadata {
val UnknownFilePosition = -1
class OffsetOrdering extends Ordering[LogOffsetMetadata] {
- override def compare(x: LogOffsetMetadata , y: LogOffsetMetadata ): Int = {
+ override def compare(x: LogOffsetMetadata, y: LogOffsetMetadata): Int = {
x.offsetDiff(y).toInt
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
index d5d7a13..3811be3 100644
--- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
+++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
@@ -246,7 +246,7 @@ class ReplicaFetcherThread(name: String,
throw new SocketTimeoutException(s"Failed to connect within $socketTimeout ms")
else {
val clientRequest = new ClientRequest(sourceBroker.id.toString, time.milliseconds(), true, header, request, null)
- networkClient.blockingSendAndReceive(clientRequest, request)(time)
+ networkClient.blockingSendAndReceive(clientRequest)(time)
}
}
catch {
@@ -260,7 +260,7 @@ class ReplicaFetcherThread(name: String,
private def earliestOrLatestOffset(topicPartition: TopicPartition, earliestOrLatest: Long, consumerId: Int): Long = {
val (request, apiVersion) =
if (brokerConfig.interBrokerProtocolVersion >= KAFKA_0_10_1_IV2) {
- val partitions = Map(topicPartition -> java.lang.Long.valueOf(earliestOrLatest))
+ val partitions = Map(topicPartition -> (earliestOrLatest: java.lang.Long))
(new ListOffsetRequest(partitions.asJava, consumerId), 1)
} else {
val partitions = Map(topicPartition -> new ListOffsetRequest.PartitionData(earliestOrLatest, 1))
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/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 859a7c4..87b8d90 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -26,7 +26,6 @@ import kafka.cluster.{Partition, Replica}
import kafka.common._
import kafka.controller.KafkaController
import kafka.log.{LogAppendInfo, LogManager}
-import kafka.message.InvalidMessageException
import kafka.metrics.KafkaMetricsGroup
import kafka.server.QuotaFactory.UnboundedQuota
import kafka.utils._
@@ -217,7 +216,7 @@ class ReplicaManager(val config: KafkaConfig,
def startup() {
// start ISR expiration thread
- // A follower can lag behind leader for up to config.replicaLagTimeMaxMs x (1 + 50%) before it is removed from ISR
+ // A follower can lag behind leader for up to config.replicaLagTimeMaxMs x 1.5 before it is removed from ISR
scheduler.schedule("isr-expiration", maybeShrinkIsr, period = config.replicaLagTimeMaxMs / 2, unit = TimeUnit.MILLISECONDS)
scheduler.schedule("isr-change-propagation", maybePropagateIsrChanges, period = 2500L, unit = TimeUnit.MILLISECONDS)
}
@@ -418,8 +417,6 @@ class ReplicaManager(val config: KafkaConfig,
_: RecordTooLargeException |
_: RecordBatchTooLargeException |
_: CorruptRecordException |
- _: InvalidRecordException |
- _: InvalidMessageException |
_: InvalidTimestampException) =>
(topicPartition, LogAppendResult(LogAppendInfo.UnknownLogAppendInfo, Some(e)))
case t: Throwable =>
@@ -667,7 +664,7 @@ class ReplicaManager(val config: KafkaConfig,
val partitionState = new mutable.HashMap[Partition, PartitionState]()
leaderAndISRRequest.partitionStates.asScala.foreach { case (topicPartition, stateInfo) =>
val partition = getOrCreatePartition(topicPartition)
- val partitionLeaderEpoch = partition.getLeaderEpoch()
+ val partitionLeaderEpoch = partition.getLeaderEpoch
// If the leader epoch is valid record the epoch of the controller that made the leadership decision.
// This is useful while updating the isr to maintain the decision maker controller's epoch in the zookeeper path
if (partitionLeaderEpoch < stateInfo.leaderEpoch) {
@@ -831,16 +828,16 @@ class ReplicaManager(val config: KafkaConfig,
partitionsToMakeFollower += partition
else
stateChangeLogger.info(("Broker %d skipped the become-follower state change after marking its partition as follower with correlation id %d from " +
- "controller %d epoch %d for partition [%s,%d] since the new leader %d is the same as the old leader")
+ "controller %d epoch %d for partition %s since the new leader %d is the same as the old leader")
.format(localBrokerId, correlationId, controllerId, partitionStateInfo.controllerEpoch,
- partition.topic, partition.partitionId, newLeaderBrokerId))
+ partition.topicPartition, newLeaderBrokerId))
case None =>
// The leader broker should always be present in the metadata cache.
// If not, we should record the error message and abort the transition process for this partition
stateChangeLogger.error(("Broker %d received LeaderAndIsrRequest with correlation id %d from controller" +
- " %d epoch %d for partition [%s,%d] but cannot become follower since the new leader %d is unavailable.")
+ " %d epoch %d for partition %s but cannot become follower since the new leader %d is unavailable.")
.format(localBrokerId, correlationId, controllerId, partitionStateInfo.controllerEpoch,
- partition.topic, partition.partitionId, newLeaderBrokerId))
+ partition.topicPartition, newLeaderBrokerId))
// Create the local replica even if the leader is unavailable. This is required to ensure that we include
// the partition's high watermark in the checkpoint file (see KAFKA-1647)
partition.getOrCreateReplica()
@@ -858,22 +855,22 @@ class ReplicaManager(val config: KafkaConfig,
(partition.topicPartition, partition.getOrCreateReplica().highWatermark.messageOffset)
}.toMap)
partitionsToMakeFollower.foreach { partition =>
- val topicPartitionOperationKey = new TopicPartitionOperationKey(partition.topic, partition.partitionId)
+ val topicPartitionOperationKey = new TopicPartitionOperationKey(partition.topicPartition)
tryCompleteDelayedProduce(topicPartitionOperationKey)
tryCompleteDelayedFetch(topicPartitionOperationKey)
}
partitionsToMakeFollower.foreach { partition =>
- stateChangeLogger.trace(("Broker %d truncated logs and checkpointed recovery boundaries for partition [%s,%d] as part of " +
+ stateChangeLogger.trace(("Broker %d truncated logs and checkpointed recovery boundaries for partition %s as part of " +
"become-follower request with correlation id %d from controller %d epoch %d").format(localBrokerId,
- partition.topic, partition.partitionId, correlationId, controllerId, epoch))
+ partition.topicPartition, correlationId, controllerId, epoch))
}
if (isShuttingDown.get()) {
partitionsToMakeFollower.foreach { partition =>
stateChangeLogger.trace(("Broker %d skipped the adding-fetcher step of the become-follower state change with correlation id %d from " +
- "controller %d epoch %d for partition [%s,%d] since it is shutting down").format(localBrokerId, correlationId,
- controllerId, epoch, partition.topic, partition.partitionId))
+ "controller %d epoch %d for partition %s since it is shutting down").format(localBrokerId, correlationId,
+ controllerId, epoch, partition.topicPartition))
}
}
else {
@@ -886,8 +883,8 @@ class ReplicaManager(val config: KafkaConfig,
partitionsToMakeFollower.foreach { partition =>
stateChangeLogger.trace(("Broker %d started fetcher to new leader as part of become-follower request from controller " +
- "%d epoch %d with correlation id %d for partition [%s,%d]")
- .format(localBrokerId, controllerId, epoch, correlationId, partition.topic, partition.partitionId))
+ "%d epoch %d with correlation id %d for partition %s")
+ .format(localBrokerId, controllerId, epoch, correlationId, partition.topicPartition))
}
}
} catch {
@@ -930,7 +927,7 @@ class ReplicaManager(val config: KafkaConfig,
}
private def getLeaderPartitions() : List[Partition] = {
- allPartitions.values.filter(_.leaderReplicaIfLocal().isDefined).toList
+ allPartitions.values.filter(_.leaderReplicaIfLocal.isDefined).toList
}
// Flushes the highwatermark value for all partitions to the highwatermark file
@@ -944,7 +941,7 @@ class ReplicaManager(val config: KafkaConfig,
} catch {
case e: IOException =>
fatal("Error writing to highwatermark file: ", e)
- Runtime.getRuntime().halt(1)
+ Runtime.getRuntime.halt(1)
}
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/utils/CoreUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/utils/CoreUtils.scala b/core/src/main/scala/kafka/utils/CoreUtils.scala
index 99b5aae..7fe9cc9 100755
--- a/core/src/main/scala/kafka/utils/CoreUtils.scala
+++ b/core/src/main/scala/kafka/utils/CoreUtils.scala
@@ -31,10 +31,8 @@ import org.apache.kafka.common.protocol.SecurityProtocol
import scala.collection._
import scala.collection.mutable
import kafka.cluster.EndPoint
-import org.apache.kafka.common.utils.Crc32
import org.apache.kafka.common.utils.Utils
-
/**
* General helper functions!
*
@@ -130,26 +128,6 @@ object CoreUtils extends Logging {
}
/**
- * Compute the CRC32 of the byte array
- * @param bytes The array to compute the checksum for
- * @return The CRC32
- */
- def crc32(bytes: Array[Byte]): Long = crc32(bytes, 0, bytes.length)
-
- /**
- * Compute the CRC32 of the segment of the byte array given by the specified size and offset
- * @param bytes The bytes to checksum
- * @param offset the offset at which to begin checksumming
- * @param size the number of bytes to checksum
- * @return The CRC32
- */
- def crc32(bytes: Array[Byte], offset: Int, size: Int): Long = {
- val crc = new Crc32()
- crc.update(bytes, offset, size)
- crc.getValue()
- }
-
- /**
* Read some bytes into the provided buffer, and return the number of bytes read. If the
* channel has been closed or we get -1 on the read for any reason, throw an EOFException
*/
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/utils/NetworkClientBlockingOps.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/utils/NetworkClientBlockingOps.scala b/core/src/main/scala/kafka/utils/NetworkClientBlockingOps.scala
index e3d389b..62e7d94 100644
--- a/core/src/main/scala/kafka/utils/NetworkClientBlockingOps.scala
+++ b/core/src/main/scala/kafka/utils/NetworkClientBlockingOps.scala
@@ -103,7 +103,7 @@ class NetworkClientBlockingOps(val client: NetworkClient) extends AnyVal {
* This method is useful for implementing blocking behaviour on top of the non-blocking `NetworkClient`, use it with
* care.
*/
- def blockingSendAndReceive(request: ClientRequest, body: AbstractRequest)(implicit time: Time): ClientResponse = {
+ def blockingSendAndReceive(request: ClientRequest)(implicit time: Time): ClientResponse = {
client.send(request, time.milliseconds())
pollContinuously { responses =>
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/utils/Pool.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/utils/Pool.scala b/core/src/main/scala/kafka/utils/Pool.scala
index 0cf6474..df74f29 100644
--- a/core/src/main/scala/kafka/utils/Pool.scala
+++ b/core/src/main/scala/kafka/utils/Pool.scala
@@ -33,9 +33,9 @@ class Pool[K,V](valueFactory: Option[K => V] = None) extends Iterable[(K, V)] {
m.foreach(kv => pool.put(kv._1, kv._2))
}
- def put(k: K, v: V) = pool.put(k, v)
+ def put(k: K, v: V): V = pool.put(k, v)
- def putIfNotExists(k: K, v: V) = pool.putIfAbsent(k, v)
+ def putIfNotExists(k: K, v: V): V = pool.putIfAbsent(k, v)
/**
* Gets the value associated with the given key. If there is no associated
@@ -44,27 +44,40 @@ class Pool[K,V](valueFactory: Option[K => V] = None) extends Iterable[(K, V)] {
* as lazy if its side-effects need to be avoided.
*
* @param key The key to lookup.
- * @return The final value associated with the key. This may be different from
- * the value created by the factory if another thread successfully
- * put a value.
+ * @return The final value associated with the key.
*/
- def getAndMaybePut(key: K) = {
+ def getAndMaybePut(key: K): V = {
if (valueFactory.isEmpty)
throw new KafkaException("Empty value factory in pool.")
- val curr = pool.get(key)
- if (curr == null) {
+ getAndMaybePut(key, valueFactory.get(key))
+ }
+
+ /**
+ * Gets the value associated with the given key. If there is no associated
+ * value, then create the value using the provided by `createValue` and return the
+ * value associated with the key.
+ *
+ * @param key The key to lookup.
+ * @param createValue Factory function.
+ * @return The final value associated with the key.
+ */
+ def getAndMaybePut(key: K, createValue: => V): V = {
+ val current = pool.get(key)
+ if (current == null) {
createLock synchronized {
- val curr = pool.get(key)
- if (curr == null)
- pool.put(key, valueFactory.get(key))
- pool.get(key)
+ val current = pool.get(key)
+ if (current == null) {
+ val value = createValue
+ pool.put(key, value)
+ value
+ }
+ else current
}
}
- else
- curr
+ else current
}
- def contains(id: K) = pool.containsKey(id)
+ def contains(id: K): Boolean = pool.containsKey(id)
def get(key: K): V = pool.get(key)
@@ -78,9 +91,9 @@ class Pool[K,V](valueFactory: Option[K => V] = None) extends Iterable[(K, V)] {
def clear() { pool.clear() }
- override def size = pool.size
+ override def size: Int = pool.size
- override def iterator = new Iterator[(K,V)]() {
+ override def iterator: Iterator[(K, V)] = new Iterator[(K,V)]() {
private val iter = pool.entrySet.iterator
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala
index bb93cb4..6381447 100644
--- a/core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala
@@ -15,7 +15,6 @@ package integration.kafka.api
import kafka.common.Topic
import kafka.integration.KafkaServerTestHarness
import kafka.log.Log
-import kafka.message.GZIPCompressionCodec
import kafka.server.KafkaConfig
import kafka.utils.TestUtils
import org.apache.kafka.clients.consumer.OffsetAndMetadata
@@ -27,11 +26,13 @@ import org.junit.Assert._
import scala.collection.JavaConverters._
import java.util.Properties
+import org.apache.kafka.common.record.CompressionType
+
class GroupCoordinatorIntegrationTest extends KafkaServerTestHarness {
- val offsetsTopicCompressionCodec = GZIPCompressionCodec
+ val offsetsTopicCompressionCodec = CompressionType.GZIP
val overridingProps = new Properties()
overridingProps.put(KafkaConfig.OffsetsTopicPartitionsProp, "1")
- overridingProps.put(KafkaConfig.OffsetsTopicCompressionCodecProp, offsetsTopicCompressionCodec.codec.toString)
+ overridingProps.put(KafkaConfig.OffsetsTopicCompressionCodecProp, offsetsTopicCompressionCodec.id.toString)
override def generateConfigs = TestUtils.createBrokerConfigs(1, zkConnect, enableControlledShutdown = false).map {
KafkaConfig.fromProps(_, overridingProps)
@@ -55,8 +56,8 @@ class GroupCoordinatorIntegrationTest extends KafkaServerTestHarness {
val logSegments = getGroupMetadataLogOpt.get.logSegments
val incorrectCompressionCodecs = logSegments
- .flatMap(_.log.shallowEntries.asScala.map(_.record.compressionType.id))
- .filter(_ != offsetsTopicCompressionCodec.codec)
+ .flatMap(_.log.shallowEntries.asScala.map(_.record.compressionType))
+ .filter(_ != offsetsTopicCompressionCodec)
assertEquals("Incorrect compression codecs should be empty", Seq.empty, incorrectCompressionCodecs)
consumer.close()
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala b/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
index 83280dc..ee556d7 100644
--- a/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
+++ b/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
@@ -29,8 +29,6 @@ import kafka.integration.KafkaServerTestHarness
import org.junit.{After, Before}
import scala.collection.mutable.Buffer
-import scala.util.control.Breaks.{breakable, break}
-import java.util.ConcurrentModificationException
/**
* A helper class for writing integration tests that involve producers, consumers, and servers
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/test/scala/unit/kafka/message/BaseMessageSetTestCases.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/message/BaseMessageSetTestCases.scala b/core/src/test/scala/unit/kafka/message/BaseMessageSetTestCases.scala
index ef2b0af..b7d2fa1 100644
--- a/core/src/test/scala/unit/kafka/message/BaseMessageSetTestCases.scala
+++ b/core/src/test/scala/unit/kafka/message/BaseMessageSetTestCases.scala
@@ -18,7 +18,8 @@
package kafka.message
import java.nio.ByteBuffer
-import java.nio.channels.GatheringByteChannel
+import java.nio.channels.{FileChannel, GatheringByteChannel}
+import java.nio.file.StandardOpenOption
import org.junit.Assert._
import kafka.utils.TestUtils._
@@ -118,14 +119,13 @@ trait BaseMessageSetTestCases extends JUnitSuite {
// do the write twice to ensure the message set is restored to its original state
for (_ <- 0 to 1) {
val file = tempFile()
- val fileRecords = FileRecords.open(file, true)
+ val channel = FileChannel.open(file.toPath, StandardOpenOption.READ, StandardOpenOption.WRITE)
try {
- val written = write(fileRecords.channel)
- fileRecords.resize() // resize since we wrote to the channel directly
-
+ val written = write(channel)
assertEquals("Expect to write the number of bytes in the set.", set.sizeInBytes, written)
+ val fileRecords = new FileRecords(file, channel, 0, Integer.MAX_VALUE, false)
assertEquals(set.asRecords.deepEntries.asScala.toVector, fileRecords.deepEntries.asScala.toVector)
- } finally fileRecords.close()
+ } finally channel.close()
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/test/scala/unit/kafka/server/ISRExpirationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/ISRExpirationTest.scala b/core/src/test/scala/unit/kafka/server/ISRExpirationTest.scala
index 0a03cac..348bfc3 100644
--- a/core/src/test/scala/unit/kafka/server/ISRExpirationTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ISRExpirationTest.scala
@@ -75,11 +75,11 @@ class IsrExpirationTest {
val leaderReplica = partition0.getReplica(configs.head.brokerId).get
// let the follower catch up to the Leader logEndOffset (15)
- for(replica <- partition0.assignedReplicas() - leaderReplica)
+ for (replica <- partition0.assignedReplicas - leaderReplica)
replica.updateLogReadResult(new LogReadResult(info = FetchDataInfo(new LogOffsetMetadata(15L), MemoryRecords.EMPTY),
hw = 15L,
leaderLogEndOffset = 15L,
- fetchTimeMs =time.milliseconds,
+ fetchTimeMs = time.milliseconds,
readSize = -1))
var partition0OSR = partition0.getOutOfSyncReplicas(leaderReplica, configs.head.replicaLagTimeMaxMs)
assertEquals("No replica should be out of sync", Set.empty[Int], partition0OSR.map(_.brokerId))
@@ -127,11 +127,11 @@ class IsrExpirationTest {
val leaderReplica = partition0.getReplica(configs.head.brokerId).get
// Make the remote replica not read to the end of log. It should be not be out of sync for at least 100 ms
- for(replica <- partition0.assignedReplicas() - leaderReplica)
+ for (replica <- partition0.assignedReplicas - leaderReplica)
replica.updateLogReadResult(new LogReadResult(info = FetchDataInfo(new LogOffsetMetadata(10L), MemoryRecords.EMPTY),
hw = 10L,
leaderLogEndOffset = 15L,
- fetchTimeMs =time.milliseconds,
+ fetchTimeMs = time.milliseconds,
readSize = -1))
// Simulate 2 fetch requests spanning more than 100 ms which do not read to the end of the log.
@@ -141,12 +141,13 @@ class IsrExpirationTest {
time.sleep(75)
- (partition0.assignedReplicas() - leaderReplica).foreach(
- r => r.updateLogReadResult(new LogReadResult(info = FetchDataInfo(new LogOffsetMetadata(11L), MemoryRecords.EMPTY),
- hw = 11L,
- leaderLogEndOffset = 15L,
- fetchTimeMs =time.milliseconds,
- readSize = -1)))
+ (partition0.assignedReplicas - leaderReplica).foreach { r =>
+ r.updateLogReadResult(new LogReadResult(info = FetchDataInfo(new LogOffsetMetadata(11L), MemoryRecords.EMPTY),
+ hw = 11L,
+ leaderLogEndOffset = 15L,
+ fetchTimeMs = time.milliseconds,
+ readSize = -1))
+ }
partition0OSR = partition0.getOutOfSyncReplicas(leaderReplica, configs.head.replicaLagTimeMaxMs)
assertEquals("No replica should be out of sync", Set.empty[Int], partition0OSR.map(_.brokerId))
@@ -157,12 +158,13 @@ class IsrExpirationTest {
assertEquals("Replica 1 should be out of sync", Set(configs.last.brokerId), partition0OSR.map(_.brokerId))
// Now actually make a fetch to the end of the log. The replicas should be back in ISR
- (partition0.assignedReplicas() - leaderReplica).foreach(
- r => r.updateLogReadResult(new LogReadResult(info = FetchDataInfo(new LogOffsetMetadata(15L), MemoryRecords.EMPTY),
- hw = 15L,
- leaderLogEndOffset = 15L,
- fetchTimeMs =time.milliseconds,
- readSize = -1)))
+ (partition0.assignedReplicas - leaderReplica).foreach { r =>
+ r.updateLogReadResult(new LogReadResult(info = FetchDataInfo(new LogOffsetMetadata(15L), MemoryRecords.EMPTY),
+ hw = 15L,
+ leaderLogEndOffset = 15L,
+ fetchTimeMs = time.milliseconds,
+ readSize = -1))
+ }
partition0OSR = partition0.getOutOfSyncReplicas(leaderReplica, configs.head.replicaLagTimeMaxMs)
assertEquals("No replica should be out of sync", Set.empty[Int], partition0OSR.map(_.brokerId))
@@ -180,7 +182,7 @@ class IsrExpirationTest {
// set in sync replicas for this partition to all the assigned replicas
partition.inSyncReplicas = allReplicas.toSet
// set lastCaughtUpTime to current time
- for(replica <- partition.assignedReplicas() - leaderReplica)
+ for (replica <- partition.assignedReplicas - leaderReplica)
replica.updateLogReadResult(new LogReadResult(info = FetchDataInfo(new LogOffsetMetadata(0L), MemoryRecords.EMPTY),
hw = 0L,
leaderLogEndOffset = 0L,
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/test/scala/unit/kafka/server/SimpleFetchTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/SimpleFetchTest.scala b/core/src/test/scala/unit/kafka/server/SimpleFetchTest.scala
index 1ce17dc..cffb04d 100644
--- a/core/src/test/scala/unit/kafka/server/SimpleFetchTest.scala
+++ b/core/src/test/scala/unit/kafka/server/SimpleFetchTest.scala
@@ -32,7 +32,6 @@ import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.record.{MemoryRecords, Record}
import org.easymock.EasyMock
import org.junit.Assert._
-import scala.collection.JavaConverters._
class SimpleFetchTest {
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/streams/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java
index 2d9b9a5..ac83f1c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java
@@ -219,7 +219,7 @@ public class InternalTopicIntegrationTest {
assertTrue(policies.contains(LogConfig.Compact()));
assertTrue(policies.contains(LogConfig.Delete()));
// retention should be 1 day + the window duration
- final Long retention = TimeUnit.MILLISECONDS.convert(1, TimeUnit.DAYS) + durationMs;
- assertEquals(retention, Long.valueOf(properties.getProperty(LogConfig.RetentionMsProp())));
+ final long retention = TimeUnit.MILLISECONDS.convert(1, TimeUnit.DAYS) + durationMs;
+ assertEquals(retention, Long.parseLong(properties.getProperty(LogConfig.RetentionMsProp())));
}
}
[2/2] kafka git commit: MINOR: Improvements to Record related classes
(part 1)
Posted by jg...@apache.org.
MINOR: Improvements to Record related classes (part 1)
Jason recently cleaned things up significantly by consolidating the Message/Record classes
into the common Java code in the clients module. While reviewing that, I noticed a few things
that could be improved a little more. To make reviewing easier, there will be multiple PRs.
Author: Ismael Juma <is...@juma.me.uk>
Reviewers: Ewen Cheslack-Postava <me...@ewencp.org>, Jason Gustafson <ja...@confluent.io>
Closes #2271 from ijuma/records-minor-fixes
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/6d6c77a7
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/6d6c77a7
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/6d6c77a7
Branch: refs/heads/trunk
Commit: 6d6c77a7a9c102f7508e4bc48e0d6eba1fcbc9c6
Parents: ce1cb32
Author: Ismael Juma <is...@juma.me.uk>
Authored: Tue Jan 3 19:53:20 2017 -0800
Committer: Jason Gustafson <ja...@confluent.io>
Committed: Tue Jan 3 19:53:20 2017 -0800
----------------------------------------------------------------------
.../apache/kafka/clients/MetadataUpdater.java | 4 +-
.../kafka/clients/producer/ProducerConfig.java | 4 +-
.../org/apache/kafka/common/Configurable.java | 2 +-
.../kafka/common/network/ByteBufferSend.java | 7 +-
.../apache/kafka/common/network/MultiSend.java | 40 ++---
.../kafka/common/network/NetworkSend.java | 23 ++-
.../apache/kafka/common/network/Receive.java | 6 +-
.../org/apache/kafka/common/network/Send.java | 8 +-
.../kafka/common/network/TransportLayers.java | 34 ++++
.../common/record/ByteBufferInputStream.java | 6 +-
.../common/record/ByteBufferLogInputStream.java | 12 +-
.../apache/kafka/common/record/FileRecords.java | 23 +--
.../kafka/common/record/MemoryRecords.java | 27 +++-
.../common/record/MemoryRecordsBuilder.java | 17 +-
.../org/apache/kafka/common/record/Records.java | 4 +-
.../kafka/common/record/RecordsIterator.java | 10 +-
.../kafka/common/requests/FetchRequest.java | 14 +-
.../kafka/common/requests/FetchResponse.java | 42 +++--
.../kafka/common/requests/RecordsSend.java | 15 +-
.../authenticator/SaslServerAuthenticator.java | 2 +-
.../org/apache/kafka/common/utils/Utils.java | 5 +-
.../clients/consumer/KafkaConsumerTest.java | 3 +-
.../clients/consumer/internals/FetcherTest.java | 5 +-
.../apache/kafka/common/record/RecordTest.java | 14 +-
.../common/requests/RequestResponseTest.java | 12 +-
.../main/scala/kafka/admin/AdminClient.scala | 12 +-
core/src/main/scala/kafka/api/ApiUtils.scala | 7 -
.../src/main/scala/kafka/api/FetchRequest.scala | 14 +-
.../main/scala/kafka/api/FetchResponse.scala | 138 ----------------
.../scala/kafka/api/OffsetCommitRequest.scala | 2 +-
.../main/scala/kafka/cluster/Partition.scala | 113 +++++--------
core/src/main/scala/kafka/cluster/Replica.scala | 75 ++++-----
.../scala/kafka/common/TopicAndPartition.scala | 4 +-
.../controller/ControllerChannelManager.scala | 2 +-
.../coordinator/GroupMetadataManager.scala | 2 +-
core/src/main/scala/kafka/log/LogCleaner.scala | 9 +-
core/src/main/scala/kafka/log/LogSegment.scala | 2 +-
.../src/main/scala/kafka/log/LogValidator.scala | 158 ++++++++++---------
core/src/main/scala/kafka/message/Message.scala | 16 +-
.../kafka/server/AbstractFetcherThread.scala | 2 +-
.../kafka/server/DelayedOperationKey.scala | 2 -
.../main/scala/kafka/server/KafkaServer.scala | 2 +-
.../scala/kafka/server/LogOffsetMetadata.scala | 2 +-
.../kafka/server/ReplicaFetcherThread.scala | 4 +-
.../scala/kafka/server/ReplicaManager.scala | 33 ++--
core/src/main/scala/kafka/utils/CoreUtils.scala | 22 ---
.../kafka/utils/NetworkClientBlockingOps.scala | 2 +-
core/src/main/scala/kafka/utils/Pool.scala | 47 ++++--
.../api/GroupCoordinatorIntegrationTest.scala | 11 +-
.../kafka/api/IntegrationTestHarness.scala | 2 -
.../kafka/message/BaseMessageSetTestCases.scala | 12 +-
.../unit/kafka/server/ISRExpirationTest.scala | 36 +++--
.../unit/kafka/server/SimpleFetchTest.scala | 1 -
.../InternalTopicIntegrationTest.java | 4 +-
54 files changed, 458 insertions(+), 617 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java b/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
index 37c77bb..34bdbf6 100644
--- a/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
+++ b/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
@@ -41,8 +41,8 @@ interface MetadataUpdater {
* Starts a cluster metadata update if needed and possible. Returns the time until the metadata update (which would
* be 0 if an update has been started as a result of this call).
*
- * If the implementation relies on `NetworkClient` to send requests, the completed receive will be passed to
- * `handleCompletedMetadataResponse`.
+ * If the implementation relies on `NetworkClient` to send requests, `handleCompletedMetadataResponse` will be
+ * invoked after the metadata response is received.
*
* The semantics of `needed` and `possible` are implementation-dependent and may take into account a number of
* factors like node availability, how long since the last metadata update, etc.
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
index 39446f5..73bf1c8 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
@@ -211,8 +211,8 @@ public class ProducerConfig extends AbstractConfig {
/** <code>request.timeout.ms</code> */
public static final String REQUEST_TIMEOUT_MS_CONFIG = CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG;
private static final String REQUEST_TIMEOUT_MS_DOC = CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC
- + " request.timeout.ms should be larger than replica.lag.time.max.ms, a broker side configuration,"
- + " to reduce message duplication caused by unnecessary producer retry.";
+ + " This should be larger than replica.lag.time.max.ms (a broker configuration)"
+ + " to reduce the possibility of message duplication due to unnecessary producer retries.";
/** <code>interceptor.classes</code> */
public static final String INTERCEPTOR_CLASSES_CONFIG = "interceptor.classes";
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/Configurable.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/Configurable.java b/clients/src/main/java/org/apache/kafka/common/Configurable.java
index 8d0edc0..98774cc 100644
--- a/clients/src/main/java/org/apache/kafka/common/Configurable.java
+++ b/clients/src/main/java/org/apache/kafka/common/Configurable.java
@@ -26,6 +26,6 @@ public interface Configurable {
/**
* Configure this class with the given key-value pairs
*/
- public void configure(Map<String, ?> configs);
+ void configure(Map<String, ?> configs);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java b/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java
index 3683283..a7ba6bf 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java
@@ -57,12 +57,7 @@ public class ByteBufferSend implements Send {
if (written < 0)
throw new EOFException("Wrote negative bytes to channel. This shouldn't happen.");
remaining -= written;
- // This is temporary workaround. As Send , Receive interfaces are being used by BlockingChannel.
- // Once BlockingChannel is removed we can make Send, Receive to work with transportLayer rather than
- // GatheringByteChannel or ScatteringByteChannel.
- if (channel instanceof TransportLayer)
- pending = ((TransportLayer) channel).hasPendingWrites();
-
+ pending = TransportLayers.hasPendingWrites(channel);
return written;
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/network/MultiSend.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/MultiSend.java b/clients/src/main/java/org/apache/kafka/common/network/MultiSend.java
index 11f5e07..0285fe3 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/MultiSend.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/MultiSend.java
@@ -32,19 +32,22 @@ import java.util.List;
public class MultiSend implements Send {
private static final Logger log = LoggerFactory.getLogger(MultiSend.class);
- private String dest;
+
+ private final String dest;
+ private final Iterator<Send> sendsIterator;
+ private final long size;
+
private long totalWritten = 0;
- private Iterator<Send> sendsIterator;
private Send current;
- private boolean doneSends = false;
- private long size = 0;
public MultiSend(String dest, List<Send> sends) {
this.dest = dest;
this.sendsIterator = sends.iterator();
nextSendOrDone();
- for (Send send: sends)
- this.size += send.size();
+ long size = 0;
+ for (Send send : sends)
+ size += send.size();
+ this.size = size;
}
@Override
@@ -59,40 +62,39 @@ public class MultiSend implements Send {
@Override
public boolean completed() {
- if (doneSends) {
- if (totalWritten != size)
- log.error("mismatch in sending bytes over socket; expected: " + size + " actual: " + totalWritten);
- return true;
- } else {
- return false;
- }
+ return current == null;
}
@Override
public long writeTo(GatheringByteChannel channel) throws IOException {
if (completed())
- throw new KafkaException("This operation cannot be completed on a complete request.");
+ throw new KafkaException("This operation cannot be invoked on a complete request.");
int totalWrittenPerCall = 0;
boolean sendComplete;
do {
long written = current.writeTo(channel);
- totalWritten += written;
totalWrittenPerCall += written;
sendComplete = current.completed();
if (sendComplete)
nextSendOrDone();
} while (!completed() && sendComplete);
- if (log.isTraceEnabled())
- log.trace("Bytes written as part of multisend call : " + totalWrittenPerCall + "Total bytes written so far : " + totalWritten + "Expected bytes to write : " + size);
+
+ totalWritten += totalWrittenPerCall;
+
+ if (completed() && totalWritten != size)
+ log.error("mismatch in sending bytes over socket; expected: " + size + " actual: " + totalWritten);
+
+ log.trace("Bytes written as part of multi-send call: {}, total bytes written so far: {}, expected bytes to write: {}",
+ totalWrittenPerCall, totalWritten, size);
+
return totalWrittenPerCall;
}
- // update current if there's a next Send, mark sends as done if there isn't
private void nextSendOrDone() {
if (sendsIterator.hasNext())
current = sendsIterator.next();
else
- doneSends = true;
+ current = null;
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java b/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java
index 5e4bf2c..4201579 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java
@@ -23,20 +23,19 @@ import java.nio.ByteBuffer;
*/
public class NetworkSend extends ByteBufferSend {
- public NetworkSend(String destination, ByteBuffer... buffers) {
- super(destination, sizeDelimit(buffers));
+ public NetworkSend(String destination, ByteBuffer buffer) {
+ super(destination, sizeDelimit(buffer));
}
- private static ByteBuffer[] sizeDelimit(ByteBuffer[] buffers) {
- int size = 0;
- for (ByteBuffer buffer : buffers)
- size += buffer.remaining();
- ByteBuffer[] delimited = new ByteBuffer[buffers.length + 1];
- delimited[0] = ByteBuffer.allocate(4);
- delimited[0].putInt(size);
- delimited[0].rewind();
- System.arraycopy(buffers, 0, delimited, 1, buffers.length);
- return delimited;
+ private static ByteBuffer[] sizeDelimit(ByteBuffer buffer) {
+ return new ByteBuffer[] {sizeBuffer(buffer.remaining()), buffer};
+ }
+
+ private static ByteBuffer sizeBuffer(int size) {
+ ByteBuffer sizeBuffer = ByteBuffer.allocate(4);
+ sizeBuffer.putInt(size);
+ sizeBuffer.rewind();
+ return sizeBuffer;
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/network/Receive.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Receive.java b/clients/src/main/java/org/apache/kafka/common/network/Receive.java
index 4b14431..57599f7 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Receive.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Receive.java
@@ -27,12 +27,12 @@ public interface Receive {
/**
* The numeric id of the source from which we are receiving data.
*/
- public String source();
+ String source();
/**
* Are we done receiving data?
*/
- public boolean complete();
+ boolean complete();
/**
* Read bytes into this receive from the given channel
@@ -40,6 +40,6 @@ public interface Receive {
* @return The number of bytes read
* @throws IOException If the reading fails
*/
- public long readFrom(ScatteringByteChannel channel) throws IOException;
+ long readFrom(ScatteringByteChannel channel) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/network/Send.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Send.java b/clients/src/main/java/org/apache/kafka/common/network/Send.java
index e0d8831..628fb29 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Send.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Send.java
@@ -23,12 +23,12 @@ public interface Send {
/**
* The numeric id for the destination of this send
*/
- public String destination();
+ String destination();
/**
* Is this send complete?
*/
- public boolean completed();
+ boolean completed();
/**
* Write some as-yet unwritten bytes from this send to the provided channel. It may take multiple calls for the send
@@ -37,11 +37,11 @@ public interface Send {
* @return The number of bytes written
* @throws IOException If the write fails
*/
- public long writeTo(GatheringByteChannel channel) throws IOException;
+ long writeTo(GatheringByteChannel channel) throws IOException;
/**
* Size of the send
*/
- public long size();
+ long size();
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/network/TransportLayers.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/TransportLayers.java b/clients/src/main/java/org/apache/kafka/common/network/TransportLayers.java
new file mode 100644
index 0000000..b77eae8
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/network/TransportLayers.java
@@ -0,0 +1,34 @@
+/**
+ * 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.network;
+
+import java.nio.channels.GatheringByteChannel;
+
+public final class TransportLayers {
+
+ private TransportLayers() {}
+
+ // This is temporary workaround as Send and Receive interfaces are used by BlockingChannel.
+ // Once BlockingChannel is removed we can make Send and Receive work with TransportLayer rather than
+ // GatheringByteChannel or ScatteringByteChannel.
+ public static boolean hasPendingWrites(GatheringByteChannel channel) {
+ if (channel instanceof TransportLayer)
+ return ((TransportLayer) channel).hasPendingWrites();
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/record/ByteBufferInputStream.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/ByteBufferInputStream.java b/clients/src/main/java/org/apache/kafka/common/record/ByteBufferInputStream.java
index b25f949..37e4766 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/ByteBufferInputStream.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/ByteBufferInputStream.java
@@ -23,14 +23,14 @@ import java.nio.ByteBuffer;
/**
* A byte buffer backed input inputStream
*/
-public class ByteBufferInputStream extends DataInputStream {
+public final class ByteBufferInputStream extends DataInputStream {
public ByteBufferInputStream(ByteBuffer buffer) {
super(new UnderlyingInputStream(buffer));
}
- private static class UnderlyingInputStream extends InputStream {
- private ByteBuffer buffer;
+ private static final class UnderlyingInputStream extends InputStream {
+ private final ByteBuffer buffer;
public UnderlyingInputStream(ByteBuffer buffer) {
this.buffer = buffer;
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/record/ByteBufferLogInputStream.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/ByteBufferLogInputStream.java b/clients/src/main/java/org/apache/kafka/common/record/ByteBufferLogInputStream.java
index ae0c91b..ee7d308 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/ByteBufferLogInputStream.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/ByteBufferLogInputStream.java
@@ -92,20 +92,18 @@ class ByteBufferLogInputStream implements LogInputStream<ByteBufferLogInputStrea
// We don't need to recompute crc if the timestamp is not updated.
if (record.timestampType() == TimestampType.CREATE_TIME && currentTimestamp == timestamp)
return;
-
- byte attributes = record.attributes();
- buffer.put(LOG_OVERHEAD + Record.ATTRIBUTES_OFFSET, TimestampType.CREATE_TIME.updateAttributes(attributes));
- buffer.putLong(LOG_OVERHEAD + Record.TIMESTAMP_OFFSET, timestamp);
- long crc = record.computeChecksum();
- Utils.writeUnsignedInt(buffer, LOG_OVERHEAD + Record.CRC_OFFSET, crc);
+ setTimestampAndUpdateCrc(TimestampType.CREATE_TIME, timestamp);
}
public void setLogAppendTime(long timestamp) {
if (record.magic() == Record.MAGIC_VALUE_V0)
throw new IllegalArgumentException("Cannot set timestamp for a record with magic = 0");
+ setTimestampAndUpdateCrc(TimestampType.LOG_APPEND_TIME, timestamp);
+ }
+ private void setTimestampAndUpdateCrc(TimestampType timestampType, long timestamp) {
byte attributes = record.attributes();
- buffer.put(LOG_OVERHEAD + Record.ATTRIBUTES_OFFSET, TimestampType.LOG_APPEND_TIME.updateAttributes(attributes));
+ buffer.put(LOG_OVERHEAD + Record.ATTRIBUTES_OFFSET, timestampType.updateAttributes(attributes));
buffer.putLong(LOG_OVERHEAD + Record.TIMESTAMP_OFFSET, timestamp);
long crc = record.computeChecksum();
Utils.writeUnsignedInt(buffer, LOG_OVERHEAD + Record.CRC_OFFSET, crc);
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java b/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
index 050711d..8a33dca 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
@@ -38,11 +38,8 @@ import java.util.concurrent.atomic.AtomicInteger;
*/
public class FileRecords extends AbstractRecords implements Closeable {
private final boolean isSlice;
- private final FileChannel channel;
private final int start;
private final int end;
- private volatile File file;
- private final AtomicInteger size;
private final Iterable<FileChannelLogEntry> shallowEntries;
@@ -53,6 +50,15 @@ public class FileRecords extends AbstractRecords implements Closeable {
}
};
+ // mutable state
+ private final AtomicInteger size;
+ private final FileChannel channel;
+ private volatile File file;
+
+ /**
+ * The {@code FileRecords.open} methods should be used instead of this constructor whenever possible.
+ * The constructor is visible for tests.
+ */
public FileRecords(File file,
FileChannel channel,
int start,
@@ -65,14 +71,8 @@ public class FileRecords extends AbstractRecords implements Closeable {
this.isSlice = isSlice;
this.size = new AtomicInteger();
- // set the initial size of the buffer
- resize();
-
- shallowEntries = shallowEntriesFrom(start);
- }
-
- public void resize() throws IOException {
if (isSlice) {
+ // don't check the file size if this is just a slice view
size.set(end - start);
} else {
int limit = Math.min((int) channel.size(), end);
@@ -82,6 +82,8 @@ public class FileRecords extends AbstractRecords implements Closeable {
// set the file position to the last byte in the file
channel.position(limit);
}
+
+ shallowEntries = shallowEntriesFrom(start);
}
@Override
@@ -137,6 +139,7 @@ public class FileRecords extends AbstractRecords implements Closeable {
throw new IllegalArgumentException("Invalid size: " + size);
final int end;
+ // handle integer overflow
if (this.start + position + size < 0)
end = sizeInBytes();
else
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
index 65d91c6..6c31b25 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
@@ -56,8 +56,14 @@ public class MemoryRecords extends AbstractRecords {
@Override
public long writeTo(GatheringByteChannel channel, long position, int length) throws IOException {
- ByteBuffer dup = buffer.duplicate();
+ if (position > Integer.MAX_VALUE)
+ throw new IllegalArgumentException("position should not be greater than Integer.MAX_VALUE: " + position);
+ if (position + length > buffer.limit())
+ throw new IllegalArgumentException("position+length should not be greater than buffer.limit(), position: "
+ + position + ", length: " + length + ", buffer.limit(): " + buffer.limit());
+
int pos = (int) position;
+ ByteBuffer dup = buffer.duplicate();
dup.position(pos);
dup.limit(pos + length);
return channel.write(dup);
@@ -98,10 +104,15 @@ public class MemoryRecords extends AbstractRecords {
/**
* Filter the records into the provided ByteBuffer.
* @param filter The filter function
- * @param buffer The byte buffer to write the filtered records to
+ * @param destinationBuffer The byte buffer to write the filtered records to
* @return A FilterResult with a summary of the output (for metrics)
*/
- public FilterResult filterTo(LogEntryFilter filter, ByteBuffer buffer) {
+ public FilterResult filterTo(LogEntryFilter filter, ByteBuffer destinationBuffer) {
+ return filterTo(shallowEntries(), filter, destinationBuffer);
+ }
+
+ private static FilterResult filterTo(Iterable<ByteBufferLogEntry> fromShallowEntries, LogEntryFilter filter,
+ ByteBuffer destinationBuffer) {
long maxTimestamp = Record.NO_TIMESTAMP;
long maxOffset = -1L;
long shallowOffsetOfMaxTimestamp = -1L;
@@ -110,7 +121,7 @@ public class MemoryRecords extends AbstractRecords {
int messagesRetained = 0;
int bytesRetained = 0;
- for (ByteBufferLogEntry shallowEntry : shallowEntries()) {
+ for (ByteBufferLogEntry shallowEntry : fromShallowEntries) {
bytesRead += shallowEntry.sizeInBytes();
// We use the absolute offset to decide whether to retain the message or not (this is handled by the
@@ -144,7 +155,7 @@ public class MemoryRecords extends AbstractRecords {
if (writeOriginalEntry) {
// There are no messages compacted out and no message format conversion, write the original message set back
- shallowEntry.writeTo(buffer);
+ shallowEntry.writeTo(destinationBuffer);
messagesRetained += retainedEntries.size();
bytesRetained += shallowEntry.sizeInBytes();
@@ -153,11 +164,11 @@ public class MemoryRecords extends AbstractRecords {
shallowOffsetOfMaxTimestamp = shallowEntry.offset();
}
} else if (!retainedEntries.isEmpty()) {
- ByteBuffer slice = buffer.slice();
+ ByteBuffer slice = destinationBuffer.slice();
MemoryRecordsBuilder builder = builderWithEntries(slice, shallowRecord.timestampType(), shallowRecord.compressionType(),
shallowRecord.timestamp(), retainedEntries);
MemoryRecords records = builder.build();
- buffer.position(buffer.position() + slice.position());
+ destinationBuffer.position(destinationBuffer.position() + slice.position());
messagesRetained += retainedEntries.size();
bytesRetained += records.sizeInBytes();
@@ -388,7 +399,7 @@ public class MemoryRecords extends AbstractRecords {
long logAppendTime,
List<LogEntry> entries) {
if (entries.isEmpty())
- throw new IllegalArgumentException();
+ throw new IllegalArgumentException("entries must not be empty");
LogEntry firstEntry = entries.iterator().next();
long firstOffset = firstEntry.offset();
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
index d60861b..69e9003 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
@@ -96,14 +96,15 @@ public class MemoryRecordsBuilder {
private final int writeLimit;
private final int initialCapacity;
- private MemoryRecords builtRecords;
- private long writtenUncompressed;
- private long numRecords;
- private float compressionRate;
- private long maxTimestamp;
- private long offsetOfMaxTimestamp;
+ private long writtenUncompressed = 0;
+ private long numRecords = 0;
+ private float compressionRate = 1;
+ private long maxTimestamp = Record.NO_TIMESTAMP;
+ private long offsetOfMaxTimestamp = -1;
private long lastOffset = -1;
+ private MemoryRecords builtRecords;
+
public MemoryRecordsBuilder(ByteBuffer buffer,
byte magic,
CompressionType compressionType,
@@ -117,10 +118,6 @@ public class MemoryRecordsBuilder {
this.baseOffset = baseOffset;
this.logAppendTime = logAppendTime;
this.initPos = buffer.position();
- this.numRecords = 0;
- this.writtenUncompressed = 0;
- this.compressionRate = 1;
- this.maxTimestamp = Record.NO_TIMESTAMP;
this.writeLimit = writeLimit;
this.initialCapacity = buffer.capacity();
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/record/Records.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/Records.java b/clients/src/main/java/org/apache/kafka/common/record/Records.java
index f0dbf9e..9235f92 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/Records.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/Records.java
@@ -44,11 +44,11 @@ public interface Records {
int sizeInBytes();
/**
- * Write the contents of this buffer to a channel.
+ * Attempts to write the contents of this buffer to a channel.
* @param channel The channel to write to
* @param position The position in the buffer to write from
* @param length The number of bytes to write
- * @return The number of bytes written
+ * @return The number of bytes actually written
* @throws IOException For any IO errors
*/
long writeTo(GatheringByteChannel channel, long position, int length) throws IOException;
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/record/RecordsIterator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/RecordsIterator.java b/clients/src/main/java/org/apache/kafka/common/record/RecordsIterator.java
index 4a678d5..792a857 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/RecordsIterator.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/RecordsIterator.java
@@ -34,18 +34,18 @@ import java.util.Iterator;
public class RecordsIterator extends AbstractIterator<LogEntry> {
private final boolean shallow;
private final boolean ensureMatchingMagic;
- private final int masRecordSize;
+ private final int maxRecordSize;
private final ShallowRecordsIterator<?> shallowIter;
private DeepRecordsIterator innerIter;
public RecordsIterator(LogInputStream<?> logInputStream,
boolean shallow,
boolean ensureMatchingMagic,
- int masRecordSize) {
+ int maxRecordSize) {
this.shallowIter = new ShallowRecordsIterator<>(logInputStream);
this.shallow = shallow;
this.ensureMatchingMagic = ensureMatchingMagic;
- this.masRecordSize = masRecordSize;
+ this.maxRecordSize = maxRecordSize;
}
/**
@@ -76,7 +76,7 @@ public class RecordsIterator extends AbstractIterator<LogEntry> {
// would not try to further decompress underlying messages
// There will be at least one element in the inner iterator, so we don't
// need to call hasNext() here.
- innerIter = new DeepRecordsIterator(entry, ensureMatchingMagic, masRecordSize);
+ innerIter = new DeepRecordsIterator(entry, ensureMatchingMagic, maxRecordSize);
return innerIter.next();
}
} else {
@@ -88,7 +88,7 @@ public class RecordsIterator extends AbstractIterator<LogEntry> {
return innerIter == null || !innerIter.hasNext();
}
- private static class DataLogInputStream implements LogInputStream<LogEntry> {
+ private static final class DataLogInputStream implements LogInputStream<LogEntry> {
private final DataInputStream stream;
protected final int maxMessageSize;
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
index fd4c747..b5770d4 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
@@ -182,7 +182,7 @@ public class FetchRequest extends AbstractRequest {
@Override
public AbstractResponse getErrorResponse(int versionId, Throwable e) {
- Map<TopicPartition, FetchResponse.PartitionData> responseData = new LinkedHashMap<>();
+ LinkedHashMap<TopicPartition, FetchResponse.PartitionData> responseData = new LinkedHashMap<>();
for (Map.Entry<TopicPartition, PartitionData> entry: fetchData.entrySet()) {
FetchResponse.PartitionData partitionResponse = new FetchResponse.PartitionData(Errors.forException(e).code(),
@@ -191,17 +191,7 @@ public class FetchRequest extends AbstractRequest {
responseData.put(entry.getKey(), partitionResponse);
}
- switch (versionId) {
- case 0:
- return new FetchResponse(responseData);
- case 1:
- case 2:
- case 3:
- return new FetchResponse(responseData, 0);
- default:
- throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d",
- versionId, this.getClass().getSimpleName(), ProtoUtils.latestVersion(ApiKeys.FETCH.id)));
- }
+ return new FetchResponse(versionId, responseData, 0);
}
public int replicaId() {
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
index ec2ab47..66b01a0 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
@@ -85,35 +85,29 @@ public class FetchResponse extends AbstractResponse {
}
/**
- * Constructor for Version 0
- * @param responseData fetched data grouped by topic-partition
- */
- public FetchResponse(Map<TopicPartition, PartitionData> responseData) {
- this(0, new LinkedHashMap<>(responseData), DEFAULT_THROTTLE_TIME);
- }
-
- /**
- * Constructor for Version 1 and 2
+ * Constructor for version 3.
+ *
+ * The entries in `responseData` should be in the same order as the entries in `FetchRequest.fetchData`.
+ *
* @param responseData fetched data grouped by topic-partition
* @param throttleTime Time in milliseconds the response was throttled
*/
- public FetchResponse(Map<TopicPartition, PartitionData> responseData, int throttleTime) {
- // the schema for versions 1 and 2 is the same, so we pick 2 here
- this(2, new LinkedHashMap<>(responseData), throttleTime);
+ public FetchResponse(LinkedHashMap<TopicPartition, PartitionData> responseData, int throttleTime) {
+ this(3, responseData, throttleTime);
}
/**
- * Constructor for Version 3
+ * Constructor for all versions.
+ *
+ * From version 3, the entries in `responseData` should be in the same order as the entries in
+ * `FetchRequest.fetchData`.
+ *
* @param responseData fetched data grouped by topic-partition
* @param throttleTime Time in milliseconds the response was throttled
*/
- public FetchResponse(LinkedHashMap<TopicPartition, PartitionData> responseData, int throttleTime) {
- this(3, responseData, throttleTime);
- }
-
public FetchResponse(int version, LinkedHashMap<TopicPartition, PartitionData> responseData, int throttleTime) {
- super(new Struct(ProtoUtils.responseSchema(ApiKeys.FETCH.id, version)));
- writeStruct(struct, version, responseData, throttleTime);
+ super(writeStruct(new Struct(ProtoUtils.responseSchema(ApiKeys.FETCH.id, version)), version, responseData,
+ throttleTime));
this.responseData = responseData;
this.throttleTime = throttleTime;
}
@@ -222,10 +216,10 @@ public class FetchResponse extends AbstractResponse {
sends.add(new RecordsSend(dest, records));
}
- private static void writeStruct(Struct struct,
- int version,
- LinkedHashMap<TopicPartition, PartitionData> responseData,
- int throttleTime) {
+ private static Struct writeStruct(Struct struct,
+ int version,
+ LinkedHashMap<TopicPartition, PartitionData> responseData,
+ int throttleTime) {
List<FetchRequest.TopicAndPartitionData<PartitionData>> topicsData = FetchRequest.TopicAndPartitionData.batchByTopic(responseData);
List<Struct> topicArray = new ArrayList<>();
for (FetchRequest.TopicAndPartitionData<PartitionData> topicEntry: topicsData) {
@@ -250,6 +244,8 @@ public class FetchResponse extends AbstractResponse {
if (version >= 1)
struct.set(THROTTLE_TIME_KEY_NAME, throttleTime);
+
+ return struct;
}
public static int sizeOf(int version, LinkedHashMap<TopicPartition, PartitionData> responseData) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/requests/RecordsSend.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/RecordsSend.java b/clients/src/main/java/org/apache/kafka/common/requests/RecordsSend.java
index 1c3bb0d..f1b6b25 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/RecordsSend.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/RecordsSend.java
@@ -17,7 +17,7 @@
package org.apache.kafka.common.requests;
import org.apache.kafka.common.network.Send;
-import org.apache.kafka.common.network.TransportLayer;
+import org.apache.kafka.common.network.TransportLayers;
import org.apache.kafka.common.record.Records;
import java.io.EOFException;
@@ -52,21 +52,18 @@ public class RecordsSend implements Send {
@Override
public long writeTo(GatheringByteChannel channel) throws IOException {
long written = 0;
+
if (remaining > 0) {
written = records.writeTo(channel, size() - remaining, remaining);
if (written < 0)
throw new EOFException("Wrote negative bytes to channel. This shouldn't happen.");
+ remaining -= written;
}
- if (channel instanceof TransportLayer) {
- TransportLayer tl = (TransportLayer) channel;
- pending = tl.hasPendingWrites();
-
- if (remaining <= 0 && pending)
- channel.write(EMPTY_BYTE_BUFFER);
- }
+ pending = TransportLayers.hasPendingWrites(channel);
+ if (remaining <= 0 && pending)
+ channel.write(EMPTY_BYTE_BUFFER);
- remaining -= written;
return written;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
index 223e798..27744ad 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
@@ -320,7 +320,7 @@ public class SaslServerAuthenticator implements Authenticator {
}
} catch (SchemaException | IllegalArgumentException e) {
if (saslState == SaslState.GSSAPI_OR_HANDSHAKE_REQUEST) {
- // SchemaException is thrown if the request is not in Kafka format. IIlegalArgumentException is thrown
+ // SchemaException is thrown if the request is not in Kafka format. IllegalArgumentException is thrown
// if the API key is invalid. For compatibility with 0.9.0.x where the first packet is a GSSAPI token
// starting with 0x60, revert to GSSAPI for both these exceptions.
if (LOG.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
index c5e6716..a502b32 100755
--- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -606,7 +606,6 @@ public class Utils {
return Arrays.asList(elems);
}
-
/*
* Create a string from a collection
* @param coll the collection
@@ -779,9 +778,7 @@ public class Utils {
* @param size The number of bytes to include
*/
public static long computeChecksum(ByteBuffer buffer, int start, int size) {
- Crc32 crc = new Crc32();
- crc.update(buffer.array(), buffer.arrayOffset() + start, size);
- return crc.getValue();
+ return Crc32.crc32(buffer.array(), buffer.arrayOffset() + start, size);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
index 8240d05..861dbf9 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
@@ -70,6 +70,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -1320,7 +1321,7 @@ public class KafkaConsumerTest {
}
private FetchResponse fetchResponse(Map<TopicPartition, FetchInfo> fetches) {
- Map<TopicPartition, PartitionData> tpResponses = new HashMap<>();
+ LinkedHashMap<TopicPartition, PartitionData> tpResponses = new LinkedHashMap<>();
for (Map.Entry<TopicPartition, FetchInfo> fetchEntry : fetches.entrySet()) {
TopicPartition partition = fetchEntry.getKey();
long fetchOffset = fetchEntry.getValue().offset;
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/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 0095697..706caf7 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
@@ -63,6 +63,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -753,7 +754,9 @@ public class FetcherTest {
}
private FetchResponse fetchResponse(MemoryRecords records, short error, long hw, int throttleTime) {
- return new FetchResponse(Collections.singletonMap(tp, new FetchResponse.PartitionData(error, hw, records)), throttleTime);
+ return new FetchResponse(
+ new LinkedHashMap<>(Collections.singletonMap(tp, new FetchResponse.PartitionData(error, hw, records))),
+ throttleTime);
}
private MetadataResponse newMetadataResponse(String topic, Errors error) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/test/java/org/apache/kafka/common/record/RecordTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/record/RecordTest.java b/clients/src/test/java/org/apache/kafka/common/record/RecordTest.java
index 551d820..f785c72 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/RecordTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/RecordTest.java
@@ -35,13 +35,13 @@ import org.junit.runners.Parameterized.Parameters;
@RunWith(value = Parameterized.class)
public class RecordTest {
- private byte magic;
- private long timestamp;
- private ByteBuffer key;
- private ByteBuffer value;
- private CompressionType compression;
- private TimestampType timestampType;
- private Record record;
+ private final byte magic;
+ private final long timestamp;
+ private final ByteBuffer key;
+ private final ByteBuffer value;
+ private final CompressionType compression;
+ private final TimestampType timestampType;
+ private final Record record;
public RecordTest(byte magic, long timestamp, byte[] key, byte[] value, CompressionType compression) {
this.magic = magic;
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
index a5eb22a..5f4463f 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
@@ -184,13 +184,13 @@ public class RequestResponseTest {
@Test
public void fetchResponseVersionTest() {
- Map<TopicPartition, FetchResponse.PartitionData> responseData = new HashMap<>();
+ LinkedHashMap<TopicPartition, FetchResponse.PartitionData> responseData = new LinkedHashMap<>();
MemoryRecords records = MemoryRecords.readableRecords(ByteBuffer.allocate(10));
responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData(Errors.NONE.code(), 1000000, records));
- FetchResponse v0Response = new FetchResponse(responseData);
- FetchResponse v1Response = new FetchResponse(responseData, 10);
+ FetchResponse v0Response = new FetchResponse(0, responseData, 0);
+ FetchResponse v1Response = new FetchResponse(1, responseData, 10);
assertEquals("Throttle time must be zero", 0, v0Response.getThrottleTime());
assertEquals("Throttle time must be 10", 10, v1Response.getThrottleTime());
assertEquals("Should use schema version 0", ProtoUtils.responseSchema(ApiKeys.FETCH.id, 0), v0Response.toStruct().schema());
@@ -279,7 +279,7 @@ public class RequestResponseTest {
}
private FetchResponse createFetchResponse() {
- Map<TopicPartition, FetchResponse.PartitionData> responseData = new HashMap<>();
+ LinkedHashMap<TopicPartition, FetchResponse.PartitionData> responseData = new LinkedHashMap<>();
MemoryRecords records = MemoryRecords.readableRecords(ByteBuffer.allocate(10));
responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData(Errors.NONE.code(), 1000000, records));
return new FetchResponse(responseData, 25);
@@ -580,7 +580,9 @@ public class RequestResponseTest {
private boolean closed = false;
private ByteBufferChannel(long size) {
- this.buf = ByteBuffer.allocate(Long.valueOf(size).intValue());
+ if (size > Integer.MAX_VALUE)
+ throw new IllegalArgumentException("size should be not be greater than Integer.MAX_VALUE");
+ this.buf = ByteBuffer.allocate((int) size);
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/admin/AdminClient.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AdminClient.scala b/core/src/main/scala/kafka/admin/AdminClient.scala
index a7e7ebc..3845d8d 100644
--- a/core/src/main/scala/kafka/admin/AdminClient.scala
+++ b/core/src/main/scala/kafka/admin/AdminClient.scala
@@ -53,12 +53,12 @@ class AdminClient(val time: Time,
private def sendAnyNode(api: ApiKeys, request: AbstractRequest): AbstractResponse = {
bootstrapBrokers.foreach { broker =>
- try {
- return send(broker, api, request)
- } catch {
- case e: Exception =>
- debug(s"Request $api failed against node $broker", e)
- }
+ try {
+ return send(broker, api, request)
+ } catch {
+ case e: Exception =>
+ debug(s"Request $api failed against node $broker", e)
+ }
}
throw new RuntimeException(s"Request $api failed on brokers $bootstrapBrokers")
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/api/ApiUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/ApiUtils.scala b/core/src/main/scala/kafka/api/ApiUtils.scala
index 8e8ac36..2145e8c 100644
--- a/core/src/main/scala/kafka/api/ApiUtils.scala
+++ b/core/src/main/scala/kafka/api/ApiUtils.scala
@@ -17,9 +17,7 @@
package kafka.api
import java.nio._
-import java.nio.channels.GatheringByteChannel
import kafka.common._
-import org.apache.kafka.common.network.TransportLayer
/**
* Helper functions specific to parsing or serializing requests and responses
@@ -98,10 +96,5 @@ object ApiUtils {
throw new KafkaException(name + " has value " + value + " which is not in the range " + range + ".")
else value
}
-
- private[api] def hasPendingWrites(channel: GatheringByteChannel): Boolean = channel match {
- case t: TransportLayer => t.hasPendingWrites
- case _ => false
- }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/api/FetchRequest.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/FetchRequest.scala b/core/src/main/scala/kafka/api/FetchRequest.scala
index 00897db..5e5360a 100644
--- a/core/src/main/scala/kafka/api/FetchRequest.scala
+++ b/core/src/main/scala/kafka/api/FetchRequest.scala
@@ -25,8 +25,12 @@ import kafka.network.RequestChannel
import kafka.message.MessageSet
import java.util.concurrent.atomic.AtomicInteger
import java.nio.ByteBuffer
+import java.util
+import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
+import org.apache.kafka.common.record.MemoryRecords
+import org.apache.kafka.common.requests.{FetchResponse => JFetchResponse}
import scala.collection.mutable.ArrayBuffer
import scala.util.Random
@@ -196,12 +200,14 @@ case class FetchRequest(versionId: Short = FetchRequest.CurrentVersion,
}
override def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = {
- val fetchResponsePartitionData = requestInfo.map { case (topicAndPartition, _) =>
- (topicAndPartition, FetchResponsePartitionData(Errors.forException(e).code, -1, MessageSet.Empty))
+ val responseData = new util.LinkedHashMap[TopicPartition, JFetchResponse.PartitionData]
+ requestInfo.foreach { case (TopicAndPartition(topic, partition), _) =>
+ responseData.put(new TopicPartition(topic, partition),
+ new JFetchResponse.PartitionData(Errors.forException(e).code, -1, MemoryRecords.EMPTY))
}
- val errorResponse = FetchResponse(correlationId, fetchResponsePartitionData, request.header.apiVersion)
+ val errorResponse = new JFetchResponse(versionId, responseData, 0)
// Magic value does not matter here because the message set is empty
- requestChannel.sendResponse(new RequestChannel.Response(request, new FetchResponseSend(request.connectionId, errorResponse)))
+ requestChannel.sendResponse(new RequestChannel.Response(request, errorResponse))
}
override def describe(details: Boolean): String = {
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/api/FetchResponse.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/FetchResponse.scala b/core/src/main/scala/kafka/api/FetchResponse.scala
index 5e2a999..cd346cb 100644
--- a/core/src/main/scala/kafka/api/FetchResponse.scala
+++ b/core/src/main/scala/kafka/api/FetchResponse.scala
@@ -18,17 +18,13 @@
package kafka.api
import java.nio.ByteBuffer
-import java.nio.channels.GatheringByteChannel
import kafka.common.TopicAndPartition
import kafka.message.{ByteBufferMessageSet, MessageSet}
import kafka.api.ApiUtils._
-import org.apache.kafka.common.KafkaException
-import org.apache.kafka.common.network.{MultiSend, Send}
import org.apache.kafka.common.protocol.Errors
import scala.collection._
-import scala.collection.JavaConverters._
object FetchResponsePartitionData {
def readFrom(buffer: ByteBuffer): FetchResponsePartitionData = {
@@ -51,48 +47,6 @@ case class FetchResponsePartitionData(error: Short = Errors.NONE.code, hw: Long
val sizeInBytes = FetchResponsePartitionData.headerSize + messages.sizeInBytes
}
-// SENDS
-
-class PartitionDataSend(val partitionId: Int,
- val partitionData: FetchResponsePartitionData) extends Send {
- private val emptyBuffer = ByteBuffer.allocate(0)
- private val messageSize = partitionData.messages.sizeInBytes
- private var messagesSentSize = 0
- private var pending = false
- private val buffer = ByteBuffer.allocate( 4 /** partitionId **/ + FetchResponsePartitionData.headerSize)
- buffer.putInt(partitionId)
- buffer.putShort(partitionData.error)
- buffer.putLong(partitionData.hw)
- buffer.putInt(partitionData.messages.sizeInBytes)
- buffer.rewind()
-
- override def completed = !buffer.hasRemaining && messagesSentSize >= messageSize && !pending
-
- override def destination: String = ""
-
- override def writeTo(channel: GatheringByteChannel): Long = {
- var written = 0L
- if (buffer.hasRemaining)
- written += channel.write(buffer)
- if (!buffer.hasRemaining) {
- if (messagesSentSize < messageSize) {
- val records = partitionData.messages.asRecords
- val bytesSent = records.writeTo(channel, messagesSentSize, messageSize - messagesSentSize).toInt
- messagesSentSize += bytesSent
- written += bytesSent
- }
- if (messagesSentSize >= messageSize && hasPendingWrites(channel))
- channel.write(emptyBuffer)
- }
-
- pending = hasPendingWrites(channel)
-
- written
- }
-
- override def size = buffer.capacity() + messageSize
-}
-
object TopicData {
def readFrom(buffer: ByteBuffer): TopicData = {
val topic = readShortString(buffer)
@@ -119,50 +73,6 @@ case class TopicData(topic: String, partitionData: Seq[(Int, FetchResponsePartit
val headerSize = TopicData.headerSize(topic)
}
-class TopicDataSend(val dest: String, val topicData: TopicData) extends Send {
-
- private val emptyBuffer = ByteBuffer.allocate(0)
-
- private var sent = 0L
-
- private var pending = false
-
- override def completed: Boolean = sent >= size && !pending
-
- override def destination: String = dest
-
- override def size = topicData.headerSize + sends.size()
-
- private val buffer = ByteBuffer.allocate(topicData.headerSize)
- writeShortString(buffer, topicData.topic)
- buffer.putInt(topicData.partitionData.size)
- buffer.rewind()
-
- private val sends = new MultiSend(dest,
- topicData.partitionData.map(d => new PartitionDataSend(d._1, d._2): Send).asJava)
-
- override def writeTo(channel: GatheringByteChannel): Long = {
- if (completed)
- throw new KafkaException("This operation cannot be completed on a complete request.")
-
- var written = 0L
- if (buffer.hasRemaining)
- written += channel.write(buffer)
- if (!buffer.hasRemaining) {
- if (!sends.completed)
- written += sends.writeTo(channel)
- if (sends.completed && hasPendingWrites(channel))
- written += channel.write(emptyBuffer)
- }
-
- pending = hasPendingWrites(channel)
-
- sent += written
- written
- }
-}
-
-
object FetchResponse {
// The request version is used to determine which fields we can expect in the response
@@ -260,51 +170,3 @@ case class FetchResponse(correlationId: Int,
def errorCode(topic: String, partition: Int) = partitionDataFor(topic, partition).error
}
-
-
-class FetchResponseSend(val dest: String, val fetchResponse: FetchResponse) extends Send {
-
- private val emptyBuffer = ByteBuffer.allocate(0)
-
- private val payloadSize = fetchResponse.sizeInBytes
-
- private var sent = 0L
-
- private var pending = false
-
- override def size = 4 /* for size byte */ + payloadSize
-
- override def completed = sent >= size && !pending
-
- override def destination = dest
-
- // The throttleTimeSize will be 0 if the request was made from a client sending a V0 style request
- private val buffer = ByteBuffer.allocate(4 /* for size */ + fetchResponse.headerSizeInBytes)
- fetchResponse.writeHeaderTo(buffer)
- buffer.rewind()
-
- private val sends = new MultiSend(dest, fetchResponse.dataGroupedByTopic.map {
- case (topic, data) => new TopicDataSend(dest, TopicData(topic, data)): Send
- }.asJava)
-
- override def writeTo(channel: GatheringByteChannel): Long = {
- if (completed)
- throw new KafkaException("This operation cannot be completed on a complete request.")
-
- var written = 0L
-
- if (buffer.hasRemaining)
- written += channel.write(buffer)
- if (!buffer.hasRemaining) {
- if (!sends.completed)
- written += sends.writeTo(channel)
- if (sends.completed && hasPendingWrites(channel))
- written += channel.write(emptyBuffer)
- }
-
- sent += written
- pending = hasPendingWrites(channel)
-
- written
- }
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/api/OffsetCommitRequest.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/OffsetCommitRequest.scala b/core/src/main/scala/kafka/api/OffsetCommitRequest.scala
index e7cd952..95663af 100644
--- a/core/src/main/scala/kafka/api/OffsetCommitRequest.scala
+++ b/core/src/main/scala/kafka/api/OffsetCommitRequest.scala
@@ -160,7 +160,7 @@ case class OffsetCommitRequest(groupId: String,
})
})
- override def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = {
+ override def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = {
val errorCode = Errors.forException(e).code
val commitStatus = requestInfo.mapValues(_ => errorCode)
val commitResponse = OffsetCommitResponse(commitStatus, correlationId)
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/cluster/Partition.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala
index 8a19f17..46815e1 100755
--- a/core/src/main/scala/kafka/cluster/Partition.scala
+++ b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -52,7 +52,7 @@ class Partition(val topic: String,
private val zkUtils = replicaManager.zkUtils
private val assignedReplicaMap = new Pool[Int, Replica]
// The read lock is only required when multiple reads are executed and needs to be in a consistent manner
- private val leaderIsrUpdateLock = new ReentrantReadWriteLock()
+ private val leaderIsrUpdateLock = new ReentrantReadWriteLock
private var zkVersion: Int = LeaderAndIsr.initialZKVersion
@volatile private var leaderEpoch: Int = LeaderAndIsr.initialLeaderEpoch - 1
@volatile var leaderReplicaIdOpt: Option[Int] = None
@@ -102,51 +102,33 @@ class Partition(val topic: String,
isLeaderReplicaLocal && inSyncReplicas.size < assignedReplicas.size
def getOrCreateReplica(replicaId: Int = localBrokerId): Replica = {
- val replicaOpt = getReplica(replicaId)
- replicaOpt match {
- case Some(replica) => replica
- case None =>
- if (isReplicaLocal(replicaId)) {
- val config = LogConfig.fromProps(logManager.defaultConfig.originals,
- AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic, topic))
- val log = logManager.createLog(topicPartition, config)
- val checkpoint = replicaManager.highWatermarkCheckpoints(log.dir.getParentFile.getAbsolutePath)
- val offsetMap = checkpoint.read
- if (!offsetMap.contains(topicPartition))
- info("No checkpointed highwatermark is found for partition [%s,%d]".format(topic, partitionId))
- val offset = offsetMap.getOrElse(topicPartition, 0L).min(log.logEndOffset)
- val localReplica = new Replica(replicaId, this, time, offset, Some(log))
- addReplicaIfNotExists(localReplica)
- } else {
- val remoteReplica = new Replica(replicaId, this, time)
- addReplicaIfNotExists(remoteReplica)
- }
- getReplica(replicaId).get
- }
+ assignedReplicaMap.getAndMaybePut(replicaId, {
+ if (isReplicaLocal(replicaId)) {
+ val config = LogConfig.fromProps(logManager.defaultConfig.originals,
+ AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic, topic))
+ val log = logManager.createLog(topicPartition, config)
+ val checkpoint = replicaManager.highWatermarkCheckpoints(log.dir.getParentFile.getAbsolutePath)
+ val offsetMap = checkpoint.read
+ if (!offsetMap.contains(topicPartition))
+ info(s"No checkpointed highwatermark is found for partition $topicPartition")
+ val offset = math.min(offsetMap.getOrElse(topicPartition, 0L), log.logEndOffset)
+ new Replica(replicaId, this, time, offset, Some(log))
+ } else new Replica(replicaId, this, time)
+ })
}
def getReplica(replicaId: Int = localBrokerId): Option[Replica] = Option(assignedReplicaMap.get(replicaId))
- def leaderReplicaIfLocal(): Option[Replica] = {
- leaderReplicaIdOpt match {
- case Some(leaderReplicaId) =>
- if (leaderReplicaId == localBrokerId)
- getReplica(localBrokerId)
- else
- None
- case None => None
- }
- }
+ def leaderReplicaIfLocal: Option[Replica] =
+ leaderReplicaIdOpt.filter(_ == localBrokerId).flatMap(getReplica)
- def addReplicaIfNotExists(replica: Replica) = {
+ def addReplicaIfNotExists(replica: Replica): Replica =
assignedReplicaMap.putIfNotExists(replica.brokerId, replica)
- }
- def assignedReplicas(): Set[Replica] = {
+ def assignedReplicas: Set[Replica] =
assignedReplicaMap.values.toSet
- }
- def removeReplica(replicaId: Int) {
+ private def removeReplica(replicaId: Int) {
assignedReplicaMap.remove(replicaId)
}
@@ -162,14 +144,12 @@ class Partition(val topic: String,
} catch {
case e: IOException =>
fatal(s"Error deleting the log for partition $topicPartition", e)
- Runtime.getRuntime().halt(1)
+ Runtime.getRuntime.halt(1)
}
}
}
- def getLeaderEpoch(): Int = {
- this.leaderEpoch
- }
+ def getLeaderEpoch: Int = this.leaderEpoch
/**
* Make the local replica the leader by resetting LogEndOffset for remote replicas (there could be old LogEndOffset
@@ -186,7 +166,7 @@ class Partition(val topic: String,
allReplicas.foreach(replica => getOrCreateReplica(replica))
val newInSyncReplicas = partitionStateInfo.isr.asScala.map(r => getOrCreateReplica(r)).toSet
// remove assigned replicas that have been removed by the controller
- (assignedReplicas().map(_.brokerId) -- allReplicas).foreach(removeReplica)
+ (assignedReplicas.map(_.brokerId) -- allReplicas).foreach(removeReplica)
inSyncReplicas = newInSyncReplicas
leaderEpoch = partitionStateInfo.leaderEpoch
zkVersion = partitionStateInfo.zkVersion
@@ -201,7 +181,7 @@ class Partition(val topic: String,
val curLeaderLogEndOffset = leaderReplica.logEndOffset.messageOffset
val curTimeMs = time.milliseconds
// initialize lastCaughtUpTime of replicas as well as their lastFetchTimeMs and lastFetchLeaderLogEndOffset.
- (assignedReplicas() - leaderReplica).foreach{replica =>
+ (assignedReplicas - leaderReplica).foreach { replica =>
val lastCaughtUpTimeMs = if (inSyncReplicas.contains(replica)) curTimeMs else 0L
replica.resetLastCaughtUpTime(curLeaderLogEndOffset, curTimeMs, lastCaughtUpTimeMs)
}
@@ -234,7 +214,7 @@ class Partition(val topic: String,
// add replicas that are new
allReplicas.foreach(r => getOrCreateReplica(r))
// remove assigned replicas that have been removed by the controller
- (assignedReplicas().map(_.brokerId) -- allReplicas).foreach(removeReplica(_))
+ (assignedReplicas.map(_.brokerId) -- allReplicas).foreach(removeReplica)
inSyncReplicas = Set.empty[Replica]
leaderEpoch = partitionStateInfo.leaderEpoch
zkVersion = partitionStateInfo.zkVersion
@@ -268,7 +248,7 @@ class Partition(val topic: String,
.format(localBrokerId,
replicaId,
logReadResult.info.fetchOffsetMetadata.messageOffset,
- assignedReplicas().map(_.brokerId).mkString(","),
+ assignedReplicas.map(_.brokerId).mkString(","),
topicPartition))
}
}
@@ -286,7 +266,7 @@ class Partition(val topic: String,
def maybeExpandIsr(replicaId: Int, logReadResult: LogReadResult) {
val leaderHWIncremented = inWriteLock(leaderIsrUpdateLock) {
// check if this replica needs to be added to the ISR
- leaderReplicaIfLocal() match {
+ leaderReplicaIfLocal match {
case Some(leaderReplica) =>
val replica = getReplica(replicaId).get
val leaderHW = leaderReplica.highWatermark
@@ -294,9 +274,8 @@ class Partition(val topic: String,
assignedReplicas.map(_.brokerId).contains(replicaId) &&
replica.logEndOffset.offsetDiff(leaderHW) >= 0) {
val newInSyncReplicas = inSyncReplicas + replica
- info("Expanding ISR for partition [%s,%d] from %s to %s"
- .format(topic, partitionId, inSyncReplicas.map(_.brokerId).mkString(","),
- newInSyncReplicas.map(_.brokerId).mkString(",")))
+ info(s"Expanding ISR for partition $topicPartition from ${inSyncReplicas.map(_.brokerId).mkString(",")} " +
+ s"to ${newInSyncReplicas.map(_.brokerId).mkString(",")}")
// update ISR in ZK and cache
updateIsr(newInSyncReplicas)
replicaManager.isrExpandRate.mark()
@@ -324,7 +303,7 @@ class Partition(val topic: String,
* produce request.
*/
def checkEnoughReplicasReachOffset(requiredOffset: Long): (Boolean, Errors) = {
- leaderReplicaIfLocal() match {
+ leaderReplicaIfLocal match {
case Some(leaderReplica) =>
// keep the current immutable replica list reference
val curInSyncReplicas = inSyncReplicas
@@ -380,7 +359,7 @@ class Partition(val topic: String,
* since all callers of this private API acquire that lock
*/
private def maybeIncrementLeaderHW(leaderReplica: Replica, curTime: Long = time.milliseconds): Boolean = {
- val allLogEndOffsets = assignedReplicas.filter{replica =>
+ val allLogEndOffsets = assignedReplicas.filter { replica =>
curTime - replica.lastCaughtUpTimeMs <= replicaManager.config.replicaLagTimeMaxMs || inSyncReplicas.contains(replica)
}.map(_.logEndOffset)
val newHighWatermark = allLogEndOffsets.min(new LogOffsetMetadata.OffsetOrdering)
@@ -400,14 +379,14 @@ class Partition(val topic: String,
* Try to complete any pending requests. This should be called without holding the leaderIsrUpdateLock.
*/
private def tryCompleteDelayedRequests() {
- val requestKey = new TopicPartitionOperationKey(this.topic, this.partitionId)
+ val requestKey = new TopicPartitionOperationKey(topicPartition)
replicaManager.tryCompleteDelayedFetch(requestKey)
replicaManager.tryCompleteDelayedProduce(requestKey)
}
def maybeShrinkIsr(replicaMaxLagTimeMs: Long) {
val leaderHWIncremented = inWriteLock(leaderIsrUpdateLock) {
- leaderReplicaIfLocal() match {
+ leaderReplicaIfLocal match {
case Some(leaderReplica) =>
val outOfSyncReplicas = getOutOfSyncReplicas(leaderReplica, replicaMaxLagTimeMs)
if(outOfSyncReplicas.nonEmpty) {
@@ -457,8 +436,7 @@ class Partition(val topic: String,
def appendRecordsToLeader(records: MemoryRecords, requiredAcks: Int = 0) = {
val (info, leaderHWIncremented) = inReadLock(leaderIsrUpdateLock) {
- val leaderReplicaOpt = leaderReplicaIfLocal()
- leaderReplicaOpt match {
+ leaderReplicaIfLocal match {
case Some(leaderReplica) =>
val log = leaderReplica.log.get
val minIsr = log.config.minInSyncReplicas
@@ -466,8 +444,8 @@ class Partition(val topic: String,
// Avoid writing to leader if there are not enough insync replicas to make it safe
if (inSyncSize < minIsr && requiredAcks == -1) {
- throw new NotEnoughReplicasException("Number of insync replicas for partition [%s,%d] is [%d], below required minimum [%d]"
- .format(topic, partitionId, inSyncSize, minIsr))
+ throw new NotEnoughReplicasException("Number of insync replicas for partition %s is [%d], below required minimum [%d]"
+ .format(topicPartition, inSyncSize, minIsr))
}
val info = log.append(records, assignOffsets = true)
@@ -477,8 +455,8 @@ class Partition(val topic: String,
(info, maybeIncrementLeaderHW(leaderReplica))
case None =>
- throw new NotLeaderForPartitionException("Leader not local for partition [%s,%d] on broker %d"
- .format(topic, partitionId, localBrokerId))
+ throw new NotLeaderForPartitionException("Leader not local for partition %s on broker %d"
+ .format(topicPartition, localBrokerId))
}
}
@@ -513,18 +491,13 @@ class Partition(val topic: String,
removeMetric("ReplicasCount", tags)
}
- override def equals(that: Any): Boolean = {
- if(!that.isInstanceOf[Partition])
- return false
- val other = that.asInstanceOf[Partition]
- if(topic.equals(other.topic) && partitionId == other.partitionId)
- return true
- false
+ override def equals(that: Any): Boolean = that match {
+ case other: Partition => partitionId == other.partitionId && topic == other.topic
+ case _ => false
}
- override def hashCode(): Int = {
- 31 + topic.hashCode() + 17*partitionId
- }
+ override def hashCode: Int =
+ 31 + topic.hashCode + 17 * partitionId
override def toString: String = {
val partitionString = new StringBuilder
@@ -533,6 +506,6 @@ class Partition(val topic: String,
partitionString.append("; Leader: " + leaderReplicaIdOpt)
partitionString.append("; AssignedReplicas: " + assignedReplicaMap.keys.mkString(","))
partitionString.append("; InSyncReplicas: " + inSyncReplicas.map(_.brokerId).mkString(","))
- partitionString.toString()
+ partitionString.toString
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/cluster/Replica.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/cluster/Replica.scala b/core/src/main/scala/kafka/cluster/Replica.scala
index 4d90815..346e5d6 100644
--- a/core/src/main/scala/kafka/cluster/Replica.scala
+++ b/core/src/main/scala/kafka/cluster/Replica.scala
@@ -31,46 +31,46 @@ class Replica(val brokerId: Int,
initialHighWatermarkValue: Long = 0L,
val log: Option[Log] = None) extends Logging {
// the high watermark offset value, in non-leader replicas only its message offsets are kept
- @volatile private[this] var highWatermarkMetadata: LogOffsetMetadata = new LogOffsetMetadata(initialHighWatermarkValue)
+ @volatile private[this] var highWatermarkMetadata = new LogOffsetMetadata(initialHighWatermarkValue)
// the log end offset value, kept in all replicas;
// for local replica it is the log's end offset, for remote replicas its value is only updated by follower fetch
- @volatile private[this] var logEndOffsetMetadata: LogOffsetMetadata = LogOffsetMetadata.UnknownOffsetMetadata
+ @volatile private[this] var logEndOffsetMetadata = LogOffsetMetadata.UnknownOffsetMetadata
- // The log end offset value at the time the leader receives last FetchRequest from this follower.
+ // The log end offset value at the time the leader received the last FetchRequest from this follower
// This is used to determine the lastCaughtUpTimeMs of the follower
- @volatile private var lastFetchLeaderLogEndOffset: Long = 0L
+ @volatile private[this] var lastFetchLeaderLogEndOffset = 0L
- // The time when the leader receives last FetchRequest from this follower
+ // The time when the leader received the last FetchRequest from this follower
// This is used to determine the lastCaughtUpTimeMs of the follower
- @volatile private var lastFetchTimeMs: Long = 0L
+ @volatile private[this] var lastFetchTimeMs = 0L
- val topic = partition.topic
- val partitionId = partition.partitionId
+ // lastCaughtUpTimeMs is the largest time t such that the offset of most recent FetchRequest from this follower >=
+ // the LEO of leader at time t. This is used to determine the lag of this follower and ISR of this partition.
+ @volatile private[this] var _lastCaughtUpTimeMs = 0L
- def isLocal: Boolean = log.isDefined
+ val topicPartition = partition.topicPartition
- // lastCaughtUpTimeMs is the largest time t such that the begin offset of most recent FetchRequest from this follower >=
- // the LEO of leader at time t. This is used to determine the lag of this follower and ISR of this partition.
- private[this] val lastCaughtUpTimeMsUnderlying = new AtomicLong(0L)
+ def isLocal: Boolean = log.isDefined
- def lastCaughtUpTimeMs = lastCaughtUpTimeMsUnderlying.get()
+ def lastCaughtUpTimeMs = _lastCaughtUpTimeMs
/*
- * If the FetchRequest reads up to the log end offset of the leader when the current fetch request was received,
- * set the lastCaughtUpTimeMsUnderlying to the time when the current fetch request was received.
+ * If the FetchRequest reads up to the log end offset of the leader when the current fetch request is received,
+ * set `lastCaughtUpTimeMs` to the time when the current fetch request was received.
*
* Else if the FetchRequest reads up to the log end offset of the leader when the previous fetch request was received,
- * set the lastCaughtUpTimeMsUnderlying to the time when the previous fetch request was received.
+ * set `lastCaughtUpTimeMs` to the time when the previous fetch request was received.
*
* This is needed to enforce the semantics of ISR, i.e. a replica is in ISR if and only if it lags behind leader's LEO
- * by at most replicaLagTimeMaxMs. This semantics allows a follower to be added to the ISR even if offset of its fetch request is
- * always smaller than leader's LEO, which can happen if there are constant small produce requests at high frequency.
+ * by at most `replicaLagTimeMaxMs`. These semantics allow a follower to be added to the ISR even if the offset of its
+ * fetch request is always smaller than the leader's LEO, which can happen if small produce requests are received at
+ * high frequency.
*/
def updateLogReadResult(logReadResult : LogReadResult) {
if (logReadResult.info.fetchOffsetMetadata.messageOffset >= logReadResult.leaderLogEndOffset)
- lastCaughtUpTimeMsUnderlying.set(logReadResult.fetchTimeMs)
+ _lastCaughtUpTimeMs = logReadResult.fetchTimeMs
else if (logReadResult.info.fetchOffsetMetadata.messageOffset >= lastFetchLeaderLogEndOffset)
- lastCaughtUpTimeMsUnderlying.set(lastFetchTimeMs)
+ _lastCaughtUpTimeMs = lastFetchTimeMs
logEndOffset = logReadResult.info.fetchOffsetMetadata
lastFetchLeaderLogEndOffset = logReadResult.leaderLogEndOffset
@@ -80,16 +80,15 @@ class Replica(val brokerId: Int,
def resetLastCaughtUpTime(curLeaderLogEndOffset: Long, curTimeMs: Long, lastCaughtUpTimeMs: Long) {
lastFetchLeaderLogEndOffset = curLeaderLogEndOffset
lastFetchTimeMs = curTimeMs
- lastCaughtUpTimeMsUnderlying.set(lastCaughtUpTimeMs)
+ _lastCaughtUpTimeMs = lastCaughtUpTimeMs
}
private def logEndOffset_=(newLogEndOffset: LogOffsetMetadata) {
if (isLocal) {
- throw new KafkaException("Should not set log end offset on partition [%s,%d]'s local replica %d".format(topic, partitionId, brokerId))
+ throw new KafkaException(s"Should not set log end offset on partition $topicPartition's local replica $brokerId")
} else {
logEndOffsetMetadata = newLogEndOffset
- trace("Setting log end offset for replica %d for partition [%s,%d] to [%s]"
- .format(brokerId, topic, partitionId, logEndOffsetMetadata))
+ trace(s"Setting log end offset for replica $brokerId for partition $topicPartition to [$logEndOffsetMetadata]")
}
}
@@ -102,10 +101,9 @@ class Replica(val brokerId: Int,
def highWatermark_=(newHighWatermark: LogOffsetMetadata) {
if (isLocal) {
highWatermarkMetadata = newHighWatermark
- trace("Setting high watermark for replica %d partition [%s,%d] on broker %d to [%s]"
- .format(brokerId, topic, partitionId, brokerId, newHighWatermark))
+ trace(s"Setting high watermark for replica $brokerId partition $topicPartition to [$newHighWatermark]")
} else {
- throw new KafkaException("Should not set high watermark on partition [%s,%d]'s non-local replica %d".format(topic, partitionId, brokerId))
+ throw new KafkaException(s"Should not set high watermark on partition $topicPartition's non-local replica $brokerId")
}
}
@@ -115,31 +113,24 @@ class Replica(val brokerId: Int,
if (isLocal) {
highWatermarkMetadata = log.get.convertToOffsetMetadata(highWatermarkMetadata.messageOffset)
} else {
- throw new KafkaException("Should not construct complete high watermark on partition [%s,%d]'s non-local replica %d".format(topic, partitionId, brokerId))
+ throw new KafkaException(s"Should not construct complete high watermark on partition $topicPartition's non-local replica $brokerId")
}
}
- override def equals(that: Any): Boolean = {
- if(!that.isInstanceOf[Replica])
- return false
- val other = that.asInstanceOf[Replica]
- if(topic.equals(other.topic) && brokerId == other.brokerId && partition.equals(other.partition))
- return true
- false
- }
-
- override def hashCode(): Int = {
- 31 + topic.hashCode() + 17*brokerId + partition.hashCode()
+ override def equals(that: Any): Boolean = that match {
+ case other: Replica => brokerId == other.brokerId && topicPartition == other.topicPartition
+ case _ => false
}
+ override def hashCode: Int = 31 + topicPartition.hashCode + 17 * brokerId
override def toString: String = {
val replicaString = new StringBuilder
replicaString.append("ReplicaId: " + brokerId)
- replicaString.append("; Topic: " + topic)
+ replicaString.append("; Topic: " + partition.topic)
replicaString.append("; Partition: " + partition.partitionId)
replicaString.append("; isLocal: " + isLocal)
- if(isLocal) replicaString.append("; Highwatermark: " + highWatermark)
- replicaString.toString()
+ if (isLocal) replicaString.append("; Highwatermark: " + highWatermark)
+ replicaString.toString
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/common/TopicAndPartition.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/common/TopicAndPartition.scala b/core/src/main/scala/kafka/common/TopicAndPartition.scala
index cf892c4..35b6bcd 100644
--- a/core/src/main/scala/kafka/common/TopicAndPartition.scala
+++ b/core/src/main/scala/kafka/common/TopicAndPartition.scala
@@ -30,10 +30,10 @@ case class TopicAndPartition(topic: String, partition: Int) {
def this(partition: Partition) = this(partition.topic, partition.partitionId)
- def this(replica: Replica) = this(replica.topic, replica.partitionId)
-
def this(topicPartition: TopicPartition) = this(topicPartition.topic, topicPartition.partition)
+ def this(replica: Replica) = this(replica.topicPartition)
+
def asTuple = (topic, partition)
override def toString = "[%s,%d]".format(topic, partition)
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
index 40071b2..a820171 100755
--- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
+++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
@@ -186,7 +186,7 @@ class RequestSendThread(val controllerId: Int,
else {
val requestHeader = apiVersion.fold(networkClient.nextRequestHeader(apiKey))(networkClient.nextRequestHeader(apiKey, _))
val clientRequest = new ClientRequest(brokerNode.idString, time.milliseconds(), true, requestHeader, request, null)
- clientResponse = networkClient.blockingSendAndReceive(clientRequest, request)(time)
+ clientResponse = networkClient.blockingSendAndReceive(clientRequest)(time)
isSendSuccessful = true
}
} catch {
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala
index e649946..637b0c4 100644
--- a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala
+++ b/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala
@@ -606,7 +606,7 @@ class GroupMetadataManager(val brokerId: Int,
val partitionOpt = replicaManager.getPartition(new TopicPartition(Topic.GroupMetadataTopicName, partitionId))
val hw = partitionOpt.map { partition =>
- partition.leaderReplicaIfLocal().map(_.highWatermark.messageOffset).getOrElse(-1L)
+ partition.leaderReplicaIfLocal.map(_.highWatermark.messageOffset).getOrElse(-1L)
}.getOrElse(-1L)
hw
http://git-wip-us.apache.org/repos/asf/kafka/blob/6d6c77a7/core/src/main/scala/kafka/log/LogCleaner.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala
index 620ae9b..5f06a73 100644
--- a/core/src/main/scala/kafka/log/LogCleaner.scala
+++ b/core/src/main/scala/kafka/log/LogCleaner.scala
@@ -451,11 +451,8 @@ private[log] class Cleaner(val id: Int,
maxLogMessageSize: Int,
stats: CleanerStats) {
- def shouldRetainEntry(logEntry: LogEntry): Boolean =
- shouldRetainMessage(source, map, retainDeletes, logEntry, stats)
-
- class LogCleanerFilter extends LogEntryFilter {
- def shouldRetain(logEntry: LogEntry): Boolean = shouldRetainEntry(logEntry)
+ val logCleanerFilter = new LogEntryFilter {
+ def shouldRetain(logEntry: LogEntry): Boolean = shouldRetainMessage(source, map, retainDeletes, logEntry, stats)
}
var position = 0
@@ -468,7 +465,7 @@ private[log] class Cleaner(val id: Int,
source.log.readInto(readBuffer, position)
val records = MemoryRecords.readableRecords(readBuffer)
throttler.maybeThrottle(records.sizeInBytes)
- val result = records.filterTo(new LogCleanerFilter, writeBuffer)
+ val result = records.filterTo(logCleanerFilter, writeBuffer)
stats.readMessages(result.messagesRead, result.bytesRead)
stats.recopyMessages(result.messagesRetained, result.bytesRetained)