Index: core/src/main/scala/kafka/cluster/Replica.scala =================================================================== --- core/src/main/scala/kafka/cluster/Replica.scala (revision 1391724) +++ core/src/main/scala/kafka/cluster/Replica.scala (working copy) @@ -100,7 +100,7 @@ val replicaString = new StringBuilder replicaString.append("ReplicaId: " + brokerId) replicaString.append("; Topic: " + topic) - replicaString.append("; Partition: " + partition.toString) + replicaString.append("; Partition: " + partition.partitionId) replicaString.append("; isLocal: " + isLocal) if(isLocal) replicaString.append("; Highwatermark: " + highWatermark) replicaString.toString() Index: core/src/main/scala/kafka/cluster/Partition.scala =================================================================== --- core/src/main/scala/kafka/cluster/Partition.scala (revision 1391724) +++ core/src/main/scala/kafka/cluster/Partition.scala (working copy) @@ -192,7 +192,7 @@ case Some(leaderReplica) => val replica = getReplica(replicaId).get val leaderHW = leaderReplica.highWatermark - if (replica.logEndOffset >= leaderHW) { + if (!inSyncReplicas.contains(replica) && replica.logEndOffset >= leaderHW) { // expand ISR val newInSyncReplicas = inSyncReplicas + replica info("Expanding ISR for topic %s partition %d to %s".format(topic, partitionId, newInSyncReplicas.map(_.brokerId).mkString(","))) @@ -237,8 +237,10 @@ val allLogEndOffsets = inSyncReplicas.map(_.logEndOffset) val newHighWatermark = allLogEndOffsets.min val oldHighWatermark = leaderReplica.highWatermark - if(newHighWatermark > oldHighWatermark) + if(newHighWatermark > oldHighWatermark) { leaderReplica.highWatermark = newHighWatermark + debug("Updated highwatermark to %d".format(newHighWatermark)) + } else debug("Old hw for topic %s partition %d is %d. New hw is %d. All leo's are %s" .format(topic, partitionId, oldHighWatermark, newHighWatermark, allLogEndOffsets.mkString(","))) Index: core/src/main/scala/kafka/producer/SyncProducer.scala =================================================================== --- core/src/main/scala/kafka/producer/SyncProducer.scala (revision 1391724) +++ core/src/main/scala/kafka/producer/SyncProducer.scala (working copy) @@ -37,8 +37,7 @@ private val MaxConnectBackoffMs = 60000 private var sentOnConnection = 0 - /** make time-based reconnect starting at a random time **/ - private var lastConnectionTime = System.currentTimeMillis - SyncProducer.randomGenerator.nextDouble() * config.reconnectInterval + private var lastConnectionTime = -1L private val lock = new Object() @volatile private var shutdown: Boolean = false @@ -90,7 +89,6 @@ if(sentOnConnection >= config.reconnectInterval || (config.reconnectTimeInterval >= 0 && System.currentTimeMillis - lastConnectionTime >= config.reconnectTimeInterval)) { reconnect() sentOnConnection = 0 - lastConnectionTime = System.currentTimeMillis } response } @@ -146,6 +144,7 @@ while(!blockingChannel.isConnected && !shutdown) { try { blockingChannel.connect() + lastConnectionTime = System.currentTimeMillis info("Connected to " + config.host + ":" + config.port + " for producing") } catch { case e: Exception => { Index: core/src/main/scala/kafka/controller/KafkaController.scala =================================================================== --- core/src/main/scala/kafka/controller/KafkaController.scala (revision 1391724) +++ core/src/main/scala/kafka/controller/KafkaController.scala (working copy) @@ -116,7 +116,7 @@ updateLeaderAndIsrCache() // trigger OfflinePartition state for all partitions whose current leader is one amongst the dead brokers val partitionsWithoutLeader = controllerContext.allLeaders.filter(partitionAndLeader => - deadBrokers.contains(partitionAndLeader._2)).map(_._1).toSeq + deadBrokers.contains(partitionAndLeader._2)).keySet.toSeq partitionStateMachine.handleStateChanges(partitionsWithoutLeader, OfflinePartition) // trigger OnlinePartition state changes for offline or new partitions partitionStateMachine.triggerOnlinePartitionStateChange() Index: core/src/main/scala/kafka/controller/ControllerChannelManager.scala =================================================================== --- core/src/main/scala/kafka/controller/ControllerChannelManager.scala (revision 1391724) +++ core/src/main/scala/kafka/controller/ControllerChannelManager.scala (working copy) @@ -163,7 +163,7 @@ val broker = m._1 val leaderAndIsr = m._2 val leaderAndIsrRequest = new LeaderAndIsrRequest(leaderAndIsr) - debug(("The leaderAndIsr request sent to broker %d is %s").format(broker, leaderAndIsrRequest)) + info(("Sending to broker %d leaderAndIsr request of %s").format(broker, leaderAndIsrRequest)) sendRequest(broker, leaderAndIsrRequest, null) } brokerRequestMap.clear() Index: core/src/main/scala/kafka/message/FileMessageSet.scala =================================================================== --- core/src/main/scala/kafka/message/FileMessageSet.scala (revision 1391724) +++ core/src/main/scala/kafka/message/FileMessageSet.scala (working copy) @@ -201,6 +201,8 @@ " size of this log segment is only %d bytes".format(sizeInBytes())) channel.truncate(targetSize) setSize.set(targetSize) + /* This should not be necessary, but fixes bug 6191269 on some OSs. */ + channel.position(targetSize) } /** Index: core/src/main/scala/kafka/network/RequestChannel.scala =================================================================== --- core/src/main/scala/kafka/network/RequestChannel.scala (revision 1391724) +++ core/src/main/scala/kafka/network/RequestChannel.scala (working copy) @@ -18,15 +18,15 @@ package kafka.network import java.util.concurrent._ -import kafka.utils.SystemTime import kafka.metrics.KafkaMetricsGroup import com.yammer.metrics.core.Gauge import java.nio.ByteBuffer import kafka.api._ import kafka.common.TopicAndPartition +import kafka.utils.{Logging, SystemTime} -object RequestChannel { +object RequestChannel extends Logging { val AllDone = new Request(1, 2, getShutdownReceive(), 0) def getShutdownReceive() = { @@ -45,6 +45,7 @@ val requestId = buffer.getShort() val requestObj: RequestOrResponse = RequestKeys.deserializerForKey(requestId)(buffer) buffer.rewind() + trace("Received request: %s".format(requestObj)) def updateRequestMetrics() { val endTimeNs = SystemTime.nanoseconds @@ -70,6 +71,7 @@ m.totalTimeHist.update(totalTime) } } + trace("Completed request: %s".format(requestObj)) } case class Response(processor: Int, request: Request, responseSend: Send) { Index: core/src/main/scala/kafka/tools/DumpLogSegments.scala =================================================================== --- core/src/main/scala/kafka/tools/DumpLogSegments.scala (revision 1391724) +++ core/src/main/scala/kafka/tools/DumpLogSegments.scala (working copy) @@ -45,7 +45,10 @@ println("payload:\t" + Utils.toString(messageAndOffset.message.payload, "UTF-8")) offset = messageAndOffset.offset } - println("tail of the log is at offset: " + (startOffset + offset)) + val endOffset = startOffset + offset + println("Tail of the log is at offset: " + endOffset) + if (messageSet.sizeInBytes != endOffset) + println("Log corrupted from " + endOffset + " to " + messageSet.sizeInBytes + " is corrupted!") } } } Index: core/src/main/scala/kafka/server/ReplicaFetcherManager.scala =================================================================== --- core/src/main/scala/kafka/server/ReplicaFetcherManager.scala (revision 1391724) +++ core/src/main/scala/kafka/server/ReplicaFetcherManager.scala (working copy) @@ -23,7 +23,7 @@ extends AbstractFetcherManager("ReplicaFetcherManager on broker " + brokerConfig.brokerId + ", ", brokerConfig.numReplicaFetchers) { override def createFetcherThread(fetcherId: Int, sourceBroker: Broker): AbstractFetcherThread = { - new ReplicaFetcherThread("ReplicaFetcherThread-%d-%d-on-broker-%d-".format(sourceBroker.id, fetcherId, brokerConfig.brokerId), sourceBroker, brokerConfig, replicaMgr) + new ReplicaFetcherThread("ReplicaFetcherThread-%d-%d-on-broker-%d".format(sourceBroker.id, fetcherId, brokerConfig.brokerId), sourceBroker, brokerConfig, replicaMgr) } def shutdown() { Index: core/src/main/scala/kafka/server/KafkaApis.scala =================================================================== --- core/src/main/scala/kafka/server/KafkaApis.scala (revision 1391724) +++ core/src/main/scala/kafka/server/KafkaApis.scala (working copy) @@ -265,7 +265,7 @@ .format(partition, fetchRequest.clientId)) 0 case e => - error("Error determining available fetch bytes for topic %s partition %s on broker %s for client %s" + warn("Error determining available fetch bytes for topic %s partition %s on broker %s for client %s" .format(topic, partition, brokerId, fetchRequest.clientId), e) 0 } @@ -562,6 +562,7 @@ // unblocked if there are no partitions with pending acks val satisfied = ! partitionStatus.exists(p => p._2.acksPending) + trace("Producer request satisfaction for %s-%d = %b".format(topic, partitionId, satisfied)) satisfied } Index: core/src/main/scala/kafka/server/AbstractFetcherThread.scala =================================================================== --- core/src/main/scala/kafka/server/AbstractFetcherThread.scala (revision 1391724) +++ core/src/main/scala/kafka/server/AbstractFetcherThread.scala (working copy) @@ -112,7 +112,7 @@ warn("current offset %d for topic %s partition %d out of range; reset offset to %d" .format(currentOffset.get, topic, partitionId, newOffset)) case _ => - error("error for %s %d to broker %d".format(topic, partitionId, sourceBroker.host), + error("error for %s %d to broker %d".format(topic, partitionId, sourceBroker.id), ErrorMapping.exceptionFor(partitionData.error)) partitionsWithError += topicAndPartition fetchMap.remove(topicAndPartition)