ijuma commented on a change in pull request #9162: URL: https://github.com/apache/kafka/pull/9162#discussion_r511620727
########## File path: core/src/main/scala/kafka/log/Log.scala ########## @@ -1099,171 +1099,169 @@ class Log(@volatile private var _dir: File, val appendInfo = analyzeAndValidateRecords(records, origin, ignoreRecordSize) // return if we have no valid messages or if this is a duplicate of the last appended entry - if (appendInfo.shallowCount == 0) - return appendInfo + if (appendInfo.shallowCount == 0) appendInfo + else { - // trim any invalid bytes or partial messages before appending it to the on-disk log - var validRecords = trimInvalidBytes(records, appendInfo) + // trim any invalid bytes or partial messages before appending it to the on-disk log + var validRecords = trimInvalidBytes(records, appendInfo) - // they are valid, insert them in the log - lock synchronized { - checkIfMemoryMappedBufferClosed() - if (assignOffsets) { - // assign offsets to the message set - val offset = new LongRef(nextOffsetMetadata.messageOffset) - appendInfo.firstOffset = Some(offset.value) - val now = time.milliseconds - val validateAndOffsetAssignResult = try { - LogValidator.validateMessagesAndAssignOffsets(validRecords, - topicPartition, - offset, - time, - now, - appendInfo.sourceCodec, - appendInfo.targetCodec, - config.compact, - config.messageFormatVersion.recordVersion.value, - config.messageTimestampType, - config.messageTimestampDifferenceMaxMs, - leaderEpoch, - origin, - interBrokerProtocolVersion, - brokerTopicStats) - } catch { - case e: IOException => - throw new KafkaException(s"Error validating messages while appending to log $name", e) - } - validRecords = validateAndOffsetAssignResult.validatedRecords - appendInfo.maxTimestamp = validateAndOffsetAssignResult.maxTimestamp - appendInfo.offsetOfMaxTimestamp = validateAndOffsetAssignResult.shallowOffsetOfMaxTimestamp - appendInfo.lastOffset = offset.value - 1 - appendInfo.recordConversionStats = validateAndOffsetAssignResult.recordConversionStats - if (config.messageTimestampType == TimestampType.LOG_APPEND_TIME) - appendInfo.logAppendTime = now - - // re-validate message sizes if there's a possibility that they have changed (due to re-compression or message - // format conversion) - if (!ignoreRecordSize && validateAndOffsetAssignResult.messageSizeMaybeChanged) { - for (batch <- validRecords.batches.asScala) { - if (batch.sizeInBytes > config.maxMessageSize) { - // we record the original message set size instead of the trimmed size - // to be consistent with pre-compression bytesRejectedRate recording - brokerTopicStats.topicStats(topicPartition.topic).bytesRejectedRate.mark(records.sizeInBytes) - brokerTopicStats.allTopicsStats.bytesRejectedRate.mark(records.sizeInBytes) - throw new RecordTooLargeException(s"Message batch size is ${batch.sizeInBytes} bytes in append to" + - s"partition $topicPartition which exceeds the maximum configured size of ${config.maxMessageSize}.") - } + // they are valid, insert them in the log + lock synchronized { + checkIfMemoryMappedBufferClosed() + if (assignOffsets) { + // assign offsets to the message set + val offset = new LongRef(nextOffsetMetadata.messageOffset) + appendInfo.firstOffset = Some(offset.value) + val now = time.milliseconds + val validateAndOffsetAssignResult = try { + LogValidator.validateMessagesAndAssignOffsets(validRecords, + topicPartition, + offset, + time, + now, + appendInfo.sourceCodec, + appendInfo.targetCodec, + config.compact, + config.messageFormatVersion.recordVersion.value, + config.messageTimestampType, + config.messageTimestampDifferenceMaxMs, + leaderEpoch, + origin, + interBrokerProtocolVersion, + brokerTopicStats) + } catch { + case e: IOException => + throw new KafkaException(s"Error validating messages while appending to log $name", e) } - } - } else { - // we are taking the offsets we are given - if (!appendInfo.offsetsMonotonic) - throw new OffsetsOutOfOrderException(s"Out of order offsets found in append to $topicPartition: " + - records.records.asScala.map(_.offset)) - - if (appendInfo.firstOrLastOffsetOfFirstBatch < nextOffsetMetadata.messageOffset) { - // we may still be able to recover if the log is empty - // one example: fetching from log start offset on the leader which is not batch aligned, - // which may happen as a result of AdminClient#deleteRecords() - val firstOffset = appendInfo.firstOffset match { - case Some(offset) => offset - case None => records.batches.asScala.head.baseOffset() + validRecords = validateAndOffsetAssignResult.validatedRecords + appendInfo.maxTimestamp = validateAndOffsetAssignResult.maxTimestamp + appendInfo.offsetOfMaxTimestamp = validateAndOffsetAssignResult.shallowOffsetOfMaxTimestamp + appendInfo.lastOffset = offset.value - 1 + appendInfo.recordConversionStats = validateAndOffsetAssignResult.recordConversionStats + if (config.messageTimestampType == TimestampType.LOG_APPEND_TIME) + appendInfo.logAppendTime = now + + // re-validate message sizes if there's a possibility that they have changed (due to re-compression or message + // format conversion) + if (!ignoreRecordSize && validateAndOffsetAssignResult.messageSizeMaybeChanged) { + for (batch <- validRecords.batches.asScala) { + if (batch.sizeInBytes > config.maxMessageSize) { + // we record the original message set size instead of the trimmed size + // to be consistent with pre-compression bytesRejectedRate recording + brokerTopicStats.topicStats(topicPartition.topic).bytesRejectedRate.mark(records.sizeInBytes) + brokerTopicStats.allTopicsStats.bytesRejectedRate.mark(records.sizeInBytes) + throw new RecordTooLargeException(s"Message batch size is ${batch.sizeInBytes} bytes in append to" + + s"partition $topicPartition which exceeds the maximum configured size of ${config.maxMessageSize}.") + } + } } + } else { + // we are taking the offsets we are given + if (!appendInfo.offsetsMonotonic) + throw new OffsetsOutOfOrderException(s"Out of order offsets found in append to $topicPartition: " + + records.records.asScala.map(_.offset)) + + if (appendInfo.firstOrLastOffsetOfFirstBatch < nextOffsetMetadata.messageOffset) { + // we may still be able to recover if the log is empty + // one example: fetching from log start offset on the leader which is not batch aligned, + // which may happen as a result of AdminClient#deleteRecords() + val firstOffset = appendInfo.firstOffset match { + case Some(offset) => offset + case None => records.batches.asScala.head.baseOffset() + } - val firstOrLast = if (appendInfo.firstOffset.isDefined) "First offset" else "Last offset of the first batch" - throw new UnexpectedAppendOffsetException( - s"Unexpected offset in append to $topicPartition. $firstOrLast " + - s"${appendInfo.firstOrLastOffsetOfFirstBatch} is less than the next offset ${nextOffsetMetadata.messageOffset}. " + - s"First 10 offsets in append: ${records.records.asScala.take(10).map(_.offset)}, last offset in" + - s" append: ${appendInfo.lastOffset}. Log start offset = $logStartOffset", - firstOffset, appendInfo.lastOffset) + val firstOrLast = if (appendInfo.firstOffset.isDefined) "First offset" else "Last offset of the first batch" + throw new UnexpectedAppendOffsetException( + s"Unexpected offset in append to $topicPartition. $firstOrLast " + + s"${appendInfo.firstOrLastOffsetOfFirstBatch} is less than the next offset ${nextOffsetMetadata.messageOffset}. " + + s"First 10 offsets in append: ${records.records.asScala.take(10).map(_.offset)}, last offset in" + + s" append: ${appendInfo.lastOffset}. Log start offset = $logStartOffset", + firstOffset, appendInfo.lastOffset) + } } - } - // update the epoch cache with the epoch stamped onto the message by the leader - validRecords.batches.forEach { batch => - if (batch.magic >= RecordBatch.MAGIC_VALUE_V2) { - maybeAssignEpochStartOffset(batch.partitionLeaderEpoch, batch.baseOffset) - } else { - // In partial upgrade scenarios, we may get a temporary regression to the message format. In - // order to ensure the safety of leader election, we clear the epoch cache so that we revert - // to truncation by high watermark after the next leader election. - leaderEpochCache.filter(_.nonEmpty).foreach { cache => - warn(s"Clearing leader epoch cache after unexpected append with message format v${batch.magic}") - cache.clearAndFlush() + // update the epoch cache with the epoch stamped onto the message by the leader + validRecords.batches.forEach { batch => + if (batch.magic >= RecordBatch.MAGIC_VALUE_V2) { + maybeAssignEpochStartOffset(batch.partitionLeaderEpoch, batch.baseOffset) + } else { + // In partial upgrade scenarios, we may get a temporary regression to the message format. In + // order to ensure the safety of leader election, we clear the epoch cache so that we revert + // to truncation by high watermark after the next leader election. + leaderEpochCache.filter(_.nonEmpty).foreach { cache => + warn(s"Clearing leader epoch cache after unexpected append with message format v${batch.magic}") + cache.clearAndFlush() + } } } - } - // check messages set size may be exceed config.segmentSize - if (validRecords.sizeInBytes > config.segmentSize) { - throw new RecordBatchTooLargeException(s"Message batch size is ${validRecords.sizeInBytes} bytes in append " + - s"to partition $topicPartition, which exceeds the maximum configured segment size of ${config.segmentSize}.") - } + // check messages set size may be exceed config.segmentSize + if (validRecords.sizeInBytes > config.segmentSize) { + throw new RecordBatchTooLargeException(s"Message batch size is ${validRecords.sizeInBytes} bytes in append " + + s"to partition $topicPartition, which exceeds the maximum configured segment size of ${config.segmentSize}.") + } - // maybe roll the log if this segment is full - val segment = maybeRoll(validRecords.sizeInBytes, appendInfo) - - val logOffsetMetadata = LogOffsetMetadata( - messageOffset = appendInfo.firstOrLastOffsetOfFirstBatch, - segmentBaseOffset = segment.baseOffset, - relativePositionInSegment = segment.size) - - // now that we have valid records, offsets assigned, and timestamps updated, we need to - // validate the idempotent/transactional state of the producers and collect some metadata - val (updatedProducers, completedTxns, maybeDuplicate) = analyzeAndValidateProducerState( - logOffsetMetadata, validRecords, origin) - - maybeDuplicate.foreach { duplicate => - appendInfo.firstOffset = Some(duplicate.firstOffset) - appendInfo.lastOffset = duplicate.lastOffset - appendInfo.logAppendTime = duplicate.timestamp - appendInfo.logStartOffset = logStartOffset - return appendInfo - } + // maybe roll the log if this segment is full + val segment = maybeRoll(validRecords.sizeInBytes, appendInfo) - segment.append(largestOffset = appendInfo.lastOffset, - largestTimestamp = appendInfo.maxTimestamp, - shallowOffsetOfMaxTimestamp = appendInfo.offsetOfMaxTimestamp, - records = validRecords) - - // Increment the log end offset. We do this immediately after the append because a - // write to the transaction index below may fail and we want to ensure that the offsets - // of future appends still grow monotonically. The resulting transaction index inconsistency - // will be cleaned up after the log directory is recovered. Note that the end offset of the - // ProducerStateManager will not be updated and the last stable offset will not advance - // if the append to the transaction index fails. - updateLogEndOffset(appendInfo.lastOffset + 1) - - // update the producer state - for (producerAppendInfo <- updatedProducers.values) { - producerStateManager.update(producerAppendInfo) - } + val logOffsetMetadata = LogOffsetMetadata( + messageOffset = appendInfo.firstOrLastOffsetOfFirstBatch, + segmentBaseOffset = segment.baseOffset, + relativePositionInSegment = segment.size) - // update the transaction index with the true last stable offset. The last offset visible - // to consumers using READ_COMMITTED will be limited by this value and the high watermark. - for (completedTxn <- completedTxns) { - val lastStableOffset = producerStateManager.lastStableOffset(completedTxn) - segment.updateTxnIndex(completedTxn, lastStableOffset) - producerStateManager.completeTxn(completedTxn) - } + // now that we have valid records, offsets assigned, and timestamps updated, we need to + // validate the idempotent/transactional state of the producers and collect some metadata + val (updatedProducers, completedTxns, maybeDuplicate) = analyzeAndValidateProducerState( + logOffsetMetadata, validRecords, origin) - // always update the last producer id map offset so that the snapshot reflects the current offset - // even if there isn't any idempotent data being written - producerStateManager.updateMapEndOffset(appendInfo.lastOffset + 1) + if (maybeDuplicate.isDefined) { Review comment: Seems like we should use a pattern match here instead of `isDefined` and `get`. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org