junrao commented on a change in pull request #8103:
URL: https://github.com/apache/kafka/pull/8103#discussion_r414175405
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -20,28 +20,80 @@ package kafka.log
import java.util.Arrays
import java.security.MessageDigest
import java.nio.ByteBuffer
+
+import kafka.log.CompactionStrategy.CompactionStrategy
import kafka.utils._
-import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.record.Record
+import org.apache.kafka.common.utils.{ByteUtils, Utils}
trait OffsetMap {
+ /* The maximum number of entries this map can contain */
def slots: Int
- def put(key: ByteBuffer, offset: Long): Unit
- def get(key: ByteBuffer): Long
+
+ /* Initialize the map with the topic compact strategy */
+ def init(strategy: String, headerKey: String, cleanerThreadId: Int,
topicPartitionName: String)
Review comment:
Since this is called in every round of cleaning, perhaps it should be
called reinitialize()?
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -68,90 +120,164 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
/**
* The number of bytes of space each entry uses (the number of bytes in the
hash plus an 8 byte offset)
+ * This evaluates to the number of bytes in the hash plus 8 bytes for the
offset
Review comment:
We probably don't need the comment on the line above now?
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -20,28 +20,80 @@ package kafka.log
import java.util.Arrays
import java.security.MessageDigest
import java.nio.ByteBuffer
+
+import kafka.log.CompactionStrategy.CompactionStrategy
import kafka.utils._
-import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.record.Record
+import org.apache.kafka.common.utils.{ByteUtils, Utils}
trait OffsetMap {
+ /* The maximum number of entries this map can contain */
def slots: Int
- def put(key: ByteBuffer, offset: Long): Unit
- def get(key: ByteBuffer): Long
+
+ /* Initialize the map with the topic compact strategy */
+ def init(strategy: String, headerKey: String, cleanerThreadId: Int,
topicPartitionName: String)
+
+ /**
+ * Associate this offset to the given key.
+ * @param record The record
+ * @return success flag
+ */
+ def put(record: Record): Boolean
+
+ /**
+ * Checks to see whether to retain the record or not
+ * @param record The record
+ * @return true to retain; false not to
+ */
+ def shouldRetainRecord(record: Record): Boolean
+
+ /**
+ * Get the offset associated with this key.
+ * @param key The key
+ * @return The offset associated with this key or -1 if the key is not found
+ */
+ def getOffset(key: ByteBuffer): Long
+
+ /**
+ * Get the version associated with this key for non-offset based strategy.
+ * @param key The key
+ * @return The version associated with this key or -1 if the key is not found
+ */
+ def getVersion(key: ByteBuffer): Long
+
+ /**
+ * Sets the passed value as the latest offset.
+ * @param offset teh latest offset
+ */
def updateLatestOffset(offset: Long): Unit
- def clear(): Unit
+
+ /* The number of entries put into the map (note that not all may remain) */
Review comment:
Hmm, not sure that I understand "note that not all may remain".
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -68,90 +120,164 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
/**
* The number of bytes of space each entry uses (the number of bytes in the
hash plus an 8 byte offset)
+ * This evaluates to the number of bytes in the hash plus 8 bytes for the
offset
+ * and, if applicable, another 8 bytes for non-offset compact strategy (set
in the init method).
*/
- val bytesPerEntry = hashSize + 8
-
+ var bytesPerEntry = hashSize + longByteSize
+
/**
* The maximum number of entries this map can contain
*/
- val slots: Int = memory / bytesPerEntry
+ var slots: Int = memory / bytesPerEntry
+
+ /* compact strategy */
+ private var compactionStrategy: CompactionStrategy = null
+
+ /* header key for the Strategy header to look for */
+ private var headerKey: String = ""
+
+ /**
+ * Initialize the map with the topic compact strategy
+ * @param strategy The compaction strategy
+ * @param headerKey The header key if the compaction strategy is set to
header
+ * @param cleanerThreadId The cleaner thread id
+ * @param topicPartitionName The topic partition name
+ */
+ override def init(strategy: String = Defaults.CompactionStrategyOffset,
headerKey: String = "", cleanerThreadId: Int = -1, topicPartitionName: String =
"") {
+ // set the log indent for the topic partition
+ this.logIdent = s"[OffsetMap-$cleanerThreadId $topicPartitionName]: "
+
+ // Change the salt used for key hashing making all existing keys
unfindable.
+ this.entries = 0
+ this.lookups = 0L
+ this.probes = 0L
+ this.lastOffset = -1L
+ Arrays.fill(bytes.array, bytes.arrayOffset, bytes.arrayOffset +
bytes.limit(), 0.toByte)
+
+ this.compactionStrategy = CompactionStrategy.withName(strategy)
+ this.headerKey = headerKey.trim()
+
+ this.bytesPerEntry = hashSize + longByteSize + (if
(this.compactionStrategy == CompactionStrategy.OFFSET) 0 else longByteSize)
+ this.slots = memory / bytesPerEntry
+
+ info(s"Initialized OffsetMap with compaction strategy
'${this.compactionStrategy}' based on passed strategy '$strategy' and header
key '$headerKey'")
Review comment:
We just need to log strategy once.
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -68,90 +120,164 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
/**
* The number of bytes of space each entry uses (the number of bytes in the
hash plus an 8 byte offset)
+ * This evaluates to the number of bytes in the hash plus 8 bytes for the
offset
+ * and, if applicable, another 8 bytes for non-offset compact strategy (set
in the init method).
*/
- val bytesPerEntry = hashSize + 8
-
+ var bytesPerEntry = hashSize + longByteSize
+
/**
* The maximum number of entries this map can contain
*/
- val slots: Int = memory / bytesPerEntry
+ var slots: Int = memory / bytesPerEntry
+
+ /* compact strategy */
+ private var compactionStrategy: CompactionStrategy = null
+
+ /* header key for the Strategy header to look for */
+ private var headerKey: String = ""
+
+ /**
+ * Initialize the map with the topic compact strategy
+ * @param strategy The compaction strategy
+ * @param headerKey The header key if the compaction strategy is set to
header
+ * @param cleanerThreadId The cleaner thread id
+ * @param topicPartitionName The topic partition name
+ */
+ override def init(strategy: String = Defaults.CompactionStrategyOffset,
headerKey: String = "", cleanerThreadId: Int = -1, topicPartitionName: String =
"") {
+ // set the log indent for the topic partition
+ this.logIdent = s"[OffsetMap-$cleanerThreadId $topicPartitionName]: "
+
+ // Change the salt used for key hashing making all existing keys
unfindable.
+ this.entries = 0
+ this.lookups = 0L
+ this.probes = 0L
+ this.lastOffset = -1L
+ Arrays.fill(bytes.array, bytes.arrayOffset, bytes.arrayOffset +
bytes.limit(), 0.toByte)
+
+ this.compactionStrategy = CompactionStrategy.withName(strategy)
+ this.headerKey = headerKey.trim()
+
+ this.bytesPerEntry = hashSize + longByteSize + (if
(this.compactionStrategy == CompactionStrategy.OFFSET) 0 else longByteSize)
+ this.slots = memory / bytesPerEntry
+
+ info(s"Initialized OffsetMap with compaction strategy
'${this.compactionStrategy}' based on passed strategy '$strategy' and header
key '$headerKey'")
+ }
/**
* Associate this offset to the given key.
- * @param key The key
- * @param offset The offset
+ * @param record The record
+ * @return success flag
*/
- override def put(key: ByteBuffer, offset: Long): Unit = {
+ override def put(record: Record): Boolean = {
require(entries < slots, "Attempt to add a new entry to a full offset
map.")
+
+ val key = record.key
+ val offset = record.offset
+ val currVersion = extractVersion(record)
+
+ lastOffset = offset
lookups += 1
hashInto(key, hash1)
+
// probe until we find the first empty slot
var attempt = 0
var pos = positionOf(hash1, attempt)
- while(!isEmpty(pos)) {
+ while (!isEmpty(pos)) {
bytes.position(pos)
bytes.get(hash2)
- if(Arrays.equals(hash1, hash2)) {
+ if (Arrays.equals(hash1, hash2)) {
+ // we found an existing entry, overwrite it and return (size does not
change)
Review comment:
We need to adjust the comment since we don't always overwrite the entry.
##########
File path: core/src/main/scala/kafka/log/LogConfig.scala
##########
@@ -87,6 +90,8 @@ case class LogConfig(props: java.util.Map[_, _],
overriddenConfigs: Set[String]
val deleteRetentionMs = getLong(LogConfig.DeleteRetentionMsProp)
val compactionLagMs = getLong(LogConfig.MinCompactionLagMsProp)
val maxCompactionLagMs = getLong(LogConfig.MaxCompactionLagMsProp)
+ val compactionStrategy =
getString(LogConfig.CompactionStrategyProp).toLowerCase(Locale.ROOT)
Review comment:
It's better to convert the string to Enum here to avoid having to do the
conversion repeatedly later on.
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -68,90 +120,164 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
/**
* The number of bytes of space each entry uses (the number of bytes in the
hash plus an 8 byte offset)
+ * This evaluates to the number of bytes in the hash plus 8 bytes for the
offset
+ * and, if applicable, another 8 bytes for non-offset compact strategy (set
in the init method).
*/
- val bytesPerEntry = hashSize + 8
-
+ var bytesPerEntry = hashSize + longByteSize
+
/**
* The maximum number of entries this map can contain
*/
- val slots: Int = memory / bytesPerEntry
+ var slots: Int = memory / bytesPerEntry
+
+ /* compact strategy */
+ private var compactionStrategy: CompactionStrategy = null
+
+ /* header key for the Strategy header to look for */
+ private var headerKey: String = ""
+
+ /**
+ * Initialize the map with the topic compact strategy
+ * @param strategy The compaction strategy
+ * @param headerKey The header key if the compaction strategy is set to
header
+ * @param cleanerThreadId The cleaner thread id
+ * @param topicPartitionName The topic partition name
+ */
+ override def init(strategy: String = Defaults.CompactionStrategyOffset,
headerKey: String = "", cleanerThreadId: Int = -1, topicPartitionName: String =
"") {
+ // set the log indent for the topic partition
+ this.logIdent = s"[OffsetMap-$cleanerThreadId $topicPartitionName]: "
+
+ // Change the salt used for key hashing making all existing keys
unfindable.
+ this.entries = 0
+ this.lookups = 0L
+ this.probes = 0L
+ this.lastOffset = -1L
+ Arrays.fill(bytes.array, bytes.arrayOffset, bytes.arrayOffset +
bytes.limit(), 0.toByte)
+
+ this.compactionStrategy = CompactionStrategy.withName(strategy)
+ this.headerKey = headerKey.trim()
+
+ this.bytesPerEntry = hashSize + longByteSize + (if
(this.compactionStrategy == CompactionStrategy.OFFSET) 0 else longByteSize)
+ this.slots = memory / bytesPerEntry
+
+ info(s"Initialized OffsetMap with compaction strategy
'${this.compactionStrategy}' based on passed strategy '$strategy' and header
key '$headerKey'")
+ }
/**
* Associate this offset to the given key.
- * @param key The key
- * @param offset The offset
+ * @param record The record
+ * @return success flag
*/
- override def put(key: ByteBuffer, offset: Long): Unit = {
+ override def put(record: Record): Boolean = {
require(entries < slots, "Attempt to add a new entry to a full offset
map.")
+
+ val key = record.key
+ val offset = record.offset
+ val currVersion = extractVersion(record)
+
+ lastOffset = offset
lookups += 1
hashInto(key, hash1)
+
// probe until we find the first empty slot
var attempt = 0
var pos = positionOf(hash1, attempt)
- while(!isEmpty(pos)) {
+ while (!isEmpty(pos)) {
bytes.position(pos)
bytes.get(hash2)
- if(Arrays.equals(hash1, hash2)) {
+ if (Arrays.equals(hash1, hash2)) {
+ // we found an existing entry, overwrite it and return (size does not
change)
+ if (this.compactionStrategy != CompactionStrategy.OFFSET) {
+ // read previous value by skipping offset
+ bytes.position(bytes.position() + longByteSize)
+ val foundVersion = bytes.getLong()
+ if (foundVersion > currVersion) {
Review comment:
From the KIP, when the value is a tie, we want to keep the larger
offset. So, we need to adjust the logic here a bit.
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -197,5 +350,35 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
key.reset()
digest.digest(buffer, 0, hashSize)
}
-
+
+ /**
+ * Extract the version for the non-offset based compact strategy for
comparison
+ * @param record The record
+ * @return The version extracted from the record if the strategy is not
offset, or -1
+ */
+ private def extractVersion(record: Record): Long = {
+ if (this.compactionStrategy == CompactionStrategy.OFFSET) // offset
strategy
+ -1L
+ else if (this.compactionStrategy == CompactionStrategy.TIMESTAMP) //
record timestamp strategy
+ record.timestamp
+ else if (record.headers() == null || record.headers().isEmpty) // record
header empty
+ -1L
+ else { // header strategy
+ try {
+ // get header value if exists and positive value; otherwise return -1L
+ return record.headers()
+ .filter(it => it.value != null && it.value.nonEmpty)
Review comment:
Since `it.value `is a byte array, perhaps changing `it.value.nonEmpty`
to `it.value.length == 0` to avoid unnecessary scala conversion?
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -197,5 +350,35 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
key.reset()
digest.digest(buffer, 0, hashSize)
}
-
+
+ /**
+ * Extract the version for the non-offset based compact strategy for
comparison
+ * @param record The record
+ * @return The version extracted from the record if the strategy is not
offset, or -1
+ */
+ private def extractVersion(record: Record): Long = {
+ if (this.compactionStrategy == CompactionStrategy.OFFSET) // offset
strategy
+ -1L
Review comment:
This is not expected. So perhaps it's better to throw an
IllegalStateException?
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -68,90 +120,164 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
/**
* The number of bytes of space each entry uses (the number of bytes in the
hash plus an 8 byte offset)
+ * This evaluates to the number of bytes in the hash plus 8 bytes for the
offset
+ * and, if applicable, another 8 bytes for non-offset compact strategy (set
in the init method).
*/
- val bytesPerEntry = hashSize + 8
-
+ var bytesPerEntry = hashSize + longByteSize
+
/**
* The maximum number of entries this map can contain
*/
- val slots: Int = memory / bytesPerEntry
+ var slots: Int = memory / bytesPerEntry
+
+ /* compact strategy */
+ private var compactionStrategy: CompactionStrategy = null
+
+ /* header key for the Strategy header to look for */
+ private var headerKey: String = ""
+
+ /**
+ * Initialize the map with the topic compact strategy
+ * @param strategy The compaction strategy
+ * @param headerKey The header key if the compaction strategy is set to
header
+ * @param cleanerThreadId The cleaner thread id
+ * @param topicPartitionName The topic partition name
+ */
+ override def init(strategy: String = Defaults.CompactionStrategyOffset,
headerKey: String = "", cleanerThreadId: Int = -1, topicPartitionName: String =
"") {
+ // set the log indent for the topic partition
+ this.logIdent = s"[OffsetMap-$cleanerThreadId $topicPartitionName]: "
+
+ // Change the salt used for key hashing making all existing keys
unfindable.
+ this.entries = 0
+ this.lookups = 0L
+ this.probes = 0L
+ this.lastOffset = -1L
+ Arrays.fill(bytes.array, bytes.arrayOffset, bytes.arrayOffset +
bytes.limit(), 0.toByte)
+
+ this.compactionStrategy = CompactionStrategy.withName(strategy)
+ this.headerKey = headerKey.trim()
+
+ this.bytesPerEntry = hashSize + longByteSize + (if
(this.compactionStrategy == CompactionStrategy.OFFSET) 0 else longByteSize)
+ this.slots = memory / bytesPerEntry
+
+ info(s"Initialized OffsetMap with compaction strategy
'${this.compactionStrategy}' based on passed strategy '$strategy' and header
key '$headerKey'")
+ }
/**
* Associate this offset to the given key.
- * @param key The key
- * @param offset The offset
+ * @param record The record
+ * @return success flag
*/
- override def put(key: ByteBuffer, offset: Long): Unit = {
+ override def put(record: Record): Boolean = {
require(entries < slots, "Attempt to add a new entry to a full offset
map.")
+
+ val key = record.key
+ val offset = record.offset
+ val currVersion = extractVersion(record)
+
+ lastOffset = offset
lookups += 1
hashInto(key, hash1)
+
// probe until we find the first empty slot
var attempt = 0
var pos = positionOf(hash1, attempt)
- while(!isEmpty(pos)) {
+ while (!isEmpty(pos)) {
bytes.position(pos)
bytes.get(hash2)
- if(Arrays.equals(hash1, hash2)) {
+ if (Arrays.equals(hash1, hash2)) {
+ // we found an existing entry, overwrite it and return (size does not
change)
+ if (this.compactionStrategy != CompactionStrategy.OFFSET) {
+ // read previous value by skipping offset
+ bytes.position(bytes.position() + longByteSize)
+ val foundVersion = bytes.getLong()
+ if (foundVersion > currVersion) {
+ // map already holding latest record
+ return false
+ }
+
+ // reset the position to start of offset
+ bytes.position(bytes.position() - (2*longByteSize))
+ }
+
// we found an existing entry, overwrite it and return (size does not
change)
bytes.putLong(offset)
- lastOffset = offset
- return
+ if (this.compactionStrategy != CompactionStrategy.OFFSET) {
+ bytes.putLong(currVersion)
+ }
+
+ return true
}
+
attempt += 1
pos = positionOf(hash1, attempt)
}
+
// found an empty slot, update it--size grows by 1
bytes.position(pos)
bytes.put(hash1)
bytes.putLong(offset)
- lastOffset = offset
+ if (this.compactionStrategy != CompactionStrategy.OFFSET) {
+ bytes.putLong(currVersion)
+ }
+
entries += 1
+ true
}
-
+
/**
* Check that there is no entry at the given position
*/
private def isEmpty(position: Int): Boolean =
- bytes.getLong(position) == 0 && bytes.getLong(position + 8) == 0 &&
bytes.getLong(position + 16) == 0
+ bytes.getLong(position) == 0 && bytes.getLong(position + longByteSize) ==
0 && bytes.getLong(position + (2*longByteSize)) == 0
+
+ /**
+ * Checks to see whether to retain the record or not
+ * @param record The record
+ * @return true to retain; false not to
+ */
+ override def shouldRetainRecord(record: Record): Boolean = {
+ if (this.compactionStrategy != CompactionStrategy.OFFSET) {
+ val foundVersion = getVersion(record.key)
+ val currentVersion = extractVersion(record)
+ // use version if available & different otherwise fallback to offset
+ if (foundVersion != currentVersion)
+ return currentVersion >= foundVersion
+ }
+ val foundOffset = getOffset(record.key)
+ record.offset() >= foundOffset
+ }
/**
* Get the offset associated with this key.
* @param key The key
* @return The offset associated with this key or -1 if the key is not found
*/
- override def get(key: ByteBuffer): Long = {
- lookups += 1
- hashInto(key, hash1)
+ override def getOffset(key: ByteBuffer): Long = {
// search for the hash of this key by repeated probing until we find the
hash we are looking for or we find an empty slot
- var attempt = 0
- var pos = 0
- //we need to guard against attempt integer overflow if the map is full
- //limit attempt to number of slots once positionOf(..) enters linear
search mode
- val maxAttempts = slots + hashSize - 4
- do {
- if(attempt >= maxAttempts)
- return -1L
- pos = positionOf(hash1, attempt)
- bytes.position(pos)
- if(isEmpty(pos))
- return -1L
- bytes.get(hash2)
- attempt += 1
- } while(!Arrays.equals(hash1, hash2))
- bytes.getLong()
+ if (!search(key))
+ -1L
+ else
+ bytes.getLong()
}
-
+
/**
- * Change the salt used for key hashing making all existing keys unfindable.
+ * Get the version associated with this key for the non-offset based
strategy.
+ * @param key The key
+ * @return The version associated with this key or -1 if the key is not found
*/
- override def clear(): Unit = {
- this.entries = 0
- this.lookups = 0L
- this.probes = 0L
- this.lastOffset = -1L
- Arrays.fill(bytes.array, bytes.arrayOffset, bytes.arrayOffset +
bytes.limit(), 0.toByte)
+ override def getVersion(key: ByteBuffer): Long = {
+ if (this.compactionStrategy == CompactionStrategy.OFFSET)
+ -1L
Review comment:
This is not expected. So perhaps it's better to throw an
IllegalStateException?
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -68,90 +120,164 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
/**
* The number of bytes of space each entry uses (the number of bytes in the
hash plus an 8 byte offset)
+ * This evaluates to the number of bytes in the hash plus 8 bytes for the
offset
+ * and, if applicable, another 8 bytes for non-offset compact strategy (set
in the init method).
*/
- val bytesPerEntry = hashSize + 8
-
+ var bytesPerEntry = hashSize + longByteSize
+
/**
* The maximum number of entries this map can contain
*/
- val slots: Int = memory / bytesPerEntry
+ var slots: Int = memory / bytesPerEntry
+
+ /* compact strategy */
+ private var compactionStrategy: CompactionStrategy = null
+
+ /* header key for the Strategy header to look for */
+ private var headerKey: String = ""
+
+ /**
+ * Initialize the map with the topic compact strategy
+ * @param strategy The compaction strategy
+ * @param headerKey The header key if the compaction strategy is set to
header
+ * @param cleanerThreadId The cleaner thread id
+ * @param topicPartitionName The topic partition name
+ */
+ override def init(strategy: String = Defaults.CompactionStrategyOffset,
headerKey: String = "", cleanerThreadId: Int = -1, topicPartitionName: String =
"") {
+ // set the log indent for the topic partition
+ this.logIdent = s"[OffsetMap-$cleanerThreadId $topicPartitionName]: "
+
+ // Change the salt used for key hashing making all existing keys
unfindable.
+ this.entries = 0
+ this.lookups = 0L
+ this.probes = 0L
+ this.lastOffset = -1L
+ Arrays.fill(bytes.array, bytes.arrayOffset, bytes.arrayOffset +
bytes.limit(), 0.toByte)
+
+ this.compactionStrategy = CompactionStrategy.withName(strategy)
+ this.headerKey = headerKey.trim()
Review comment:
We probably just want to trim it once and store it in LogConfig?
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -729,6 +734,15 @@ object KafkaConfig {
val LogCleanerDeleteRetentionMsDoc = "How long are delete records retained?"
val LogCleanerMinCompactionLagMsDoc = "The minimum time a message will
remain uncompacted in the log. Only applicable for logs that are being
compacted."
val LogCleanerMaxCompactionLagMsDoc = "The maximum time a message will
remain ineligible for compaction in the log. Only applicable for logs that are
being compacted."
+ val LogCleanerCompactionStrategyDoc = "The retention strategy to use when
compacting the log. " +
+ "Only applicable for logs that are being compacted. " +
+ "By default the compaction strategy is set to \"offset\" where the latest
offset for the key is retained. " +
+ "For \"header\" strategy, the value provided by the producer in the record
header will be used to determine " +
+ "the latest record for the key. For \"timestamp\" strategy, the record
tiemstamp will be used to determine the " +
Review comment:
typo tiemstamp
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -197,5 +350,35 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
key.reset()
digest.digest(buffer, 0, hashSize)
}
-
+
+ /**
+ * Extract the version for the non-offset based compact strategy for
comparison
+ * @param record The record
+ * @return The version extracted from the record if the strategy is not
offset, or -1
+ */
+ private def extractVersion(record: Record): Long = {
+ if (this.compactionStrategy == CompactionStrategy.OFFSET) // offset
strategy
+ -1L
+ else if (this.compactionStrategy == CompactionStrategy.TIMESTAMP) //
record timestamp strategy
+ record.timestamp
+ else if (record.headers() == null || record.headers().isEmpty) // record
header empty
+ -1L
+ else { // header strategy
+ try {
+ // get header value if exists and positive value; otherwise return -1L
+ return record.headers()
+ .filter(it => it.value != null && it.value.nonEmpty)
+ .find(it => headerKey.equalsIgnoreCase(it.key.trim))
+ .map(it => ByteBuffer.wrap(it.value))
+ .map(it => ByteUtils.readVarlong(it))
+ .map(it => if (it > 0) it else -1L)
Review comment:
Assuming header value is positive is fine. Could we update the KIP?
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -197,5 +350,35 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
key.reset()
digest.digest(buffer, 0, hashSize)
}
-
+
+ /**
+ * Extract the version for the non-offset based compact strategy for
comparison
+ * @param record The record
+ * @return The version extracted from the record if the strategy is not
offset, or -1
+ */
+ private def extractVersion(record: Record): Long = {
+ if (this.compactionStrategy == CompactionStrategy.OFFSET) // offset
strategy
+ -1L
+ else if (this.compactionStrategy == CompactionStrategy.TIMESTAMP) //
record timestamp strategy
+ record.timestamp
+ else if (record.headers() == null || record.headers().isEmpty) // record
header empty
+ -1L
+ else { // header strategy
+ try {
+ // get header value if exists and positive value; otherwise return -1L
+ return record.headers()
+ .filter(it => it.value != null && it.value.nonEmpty)
+ .find(it => headerKey.equalsIgnoreCase(it.key.trim))
Review comment:
The KIP says we use the last header with duplicated header key, but the
logic here seems to take the first.
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -729,6 +734,15 @@ object KafkaConfig {
val LogCleanerDeleteRetentionMsDoc = "How long are delete records retained?"
val LogCleanerMinCompactionLagMsDoc = "The minimum time a message will
remain uncompacted in the log. Only applicable for logs that are being
compacted."
val LogCleanerMaxCompactionLagMsDoc = "The maximum time a message will
remain ineligible for compaction in the log. Only applicable for logs that are
being compacted."
+ val LogCleanerCompactionStrategyDoc = "The retention strategy to use when
compacting the log. " +
+ "Only applicable for logs that are being compacted. " +
+ "By default the compaction strategy is set to \"offset\" where the latest
offset for the key is retained. " +
+ "For \"header\" strategy, the value provided by the producer in the record
header will be used to determine " +
+ "the latest record for the key. For \"timestamp\" strategy, the record
tiemstamp will be used to determine the " +
+ "latest record for the key. So setting the strategy to anything other than
\"offset\" will replace the offset " +
+ "when calculating which records to retain for the value (i.e. provided by
the producer) matching " +
Review comment:
The part "So setting the strategy to anything other than \"offset\" will
replace the offset when calculating which records to retain for the value (i.e.
provided by the producer) matching the given strategy name (case-insensitive)."
seems unnecessary. The previous sentences explained this well already. Ditto in
TopicConfig.
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -162,9 +164,28 @@ class LogCleanerTest {
@Test
def testSizeTrimmedForPreallocatedAndCompactedTopic(): Unit = {
+ val logProps = new Properties()
+ testSizeTrimmedForPreallocatedAndCompactedTopic(logProps)
+ }
+
+ @Test
+ def
testSizeTrimmedForPreallocatedAndCompactedTopicWithTimestampCompaction(): Unit
= {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyTimestamp)
+ testSizeTrimmedForPreallocatedAndCompactedTopic(logProps)
+ }
+
+ @Test
+ def testSizeTrimmedForPreallocatedAndCompactedTopicWithHeaderCompaction():
Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyHeader)
+ logProps.put(LogConfig.CompactionStrategyHeaderKeyProp, "sequence")
+ testSizeTrimmedForPreallocatedAndCompactedTopic(logProps)
+ }
+
+ def testSizeTrimmedForPreallocatedAndCompactedTopic(logProps: Properties):
Unit = {
Review comment:
Could this be private?
##########
File path: core/src/test/scala/unit/kafka/log/FakeRecord.scala
##########
@@ -0,0 +1,74 @@
+/**
+ * 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 kafka.log
+
+import java.lang.{Long => JLong}
+import java.nio.ByteBuffer
+
+import kafka.utils.TestUtils
+import org.apache.kafka.common.header.Header
+import org.apache.kafka.common.header.internals.RecordHeader
+import org.apache.kafka.common.record.{Record, TimestampType}
+
+class FakeRecord(fakeKey: ByteBuffer, fakeOffset: Number, fakeSequence: Long =
-1L, fakeTimestamp: Long = -1L) extends Record {
+ private var fakeHeaders: Array[Header] = _
+
+ init()
+
+ private def init(): Unit = {
+ if (Option(fakeSequence).isEmpty || fakeSequence <= 0)
Review comment:
`Option(fakeSequence).isEmpty` seems unnecessary since fakeSequence is
not expected to be null.
##########
File path: core/src/test/scala/unit/kafka/log/FakeRecord.scala
##########
@@ -0,0 +1,74 @@
+/**
+ * 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 kafka.log
+
+import java.lang.{Long => JLong}
+import java.nio.ByteBuffer
+
+import kafka.utils.TestUtils
+import org.apache.kafka.common.header.Header
+import org.apache.kafka.common.header.internals.RecordHeader
+import org.apache.kafka.common.record.{Record, TimestampType}
+
+class FakeRecord(fakeKey: ByteBuffer, fakeOffset: Number, fakeSequence: Long =
-1L, fakeTimestamp: Long = -1L) extends Record {
Review comment:
Perhaps add a comment that explains this class a bit?
##########
File path: core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
##########
@@ -848,6 +850,15 @@ class KafkaConfigTest {
assertFalse(isValidKafkaConfig(props))
}
+ @Test
+ def testInvalidCompactionHeaderKeyConfig(): Unit = {
+ val props = new Properties
+ props.setProperty(KafkaConfig.LogCleanerCompactionStrategyProp, "header")
+ intercept[Exception] {
Review comment:
Could we catch InvalidConfigException?
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -251,8 +291,27 @@ class LogCleanerTest {
@Test
def testBasicTransactionAwareCleaning(): Unit = {
- val cleaner = makeCleaner(Int.MaxValue)
val logProps = new Properties()
+ testBasicTransactionAwareCleaning(logProps)
+ }
+
+ @Test
+ def testBasicTransactionAwareCleaningWithTimestampCompaction(): Unit = {
Review comment:
We need to set timestamp/header in those new tests so that we test the
new code path. Ditto in other new tests below.
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1719,6 +1738,9 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog:
Boolean, dynamicConfigO
require(logRetentionTimeMillis >= 1 || logRetentionTimeMillis == -1,
"log.retention.ms must be unlimited (-1) or, equal or greater than 1")
require(logDirs.nonEmpty, "At least one log directory must be defined via
log.dirs or log.dir.")
require(logCleanerDedupeBufferSize / logCleanerThreads > 1024 * 1024,
"log.cleaner.dedupe.buffer.size must be at least 1MB per cleaner thread.")
+ if
(Defaults.LogCleanerCompactionStrategyHeader.equalsIgnoreCase(logCleanerCompactionStrategy))
{
+ require(logCleanerCompactionStrategyHeaderKey.trim().isEmpty(),
KafkaConfig.LogCleanerCompactionStrategyHeaderKeyProp + " must be specified.")
Review comment:
It seems that we should require the headerKey to be non-empty. Do we
have a test that covers this?
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -1560,6 +2040,160 @@ class LogCleanerTest {
assertEquals("The tombstone should be retained.", 1,
log.logSegments.head.log.batches.iterator.next().lastOffset)
}
+ @Test
+ def testCleanTombstoneWithTimestampCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyTimestamp)
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a tombstone with a large timestamp.
+ log.appendAsLeader(TestUtils.singletonRecords(value = null,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0,
log.activeSegment.baseOffset))
+ // Append a message with a small timestamp and roll out a new log segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "0".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() -
logConfig.deleteRetentionMs - 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ // Append a message and roll out another log segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "1".getBytes,
+ timestamp = time.milliseconds()),
leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ }
+
+ @Test
+ def testCleanTombstoneWithHeaderCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyHeader)
+ logProps.put(LogConfig.CompactionStrategyHeaderKeyProp, "sequence")
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a message with a large timestamp.
+ var headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(1L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = "0".getBytes,
+ key = "0".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0,
log.activeSegment.baseOffset))
+ // Append a tombstone with a small timestamp and roll out a new log
segment.
+ headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(2L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = null,
+ key = "0".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds() -
logConfig.deleteRetentionMs - 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 1,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ // Append a message and roll out another log segment.
+ headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(3L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "1".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds()),
leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 1,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ }
+
+ @Test
+ def testTimestampCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyTimestamp)
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a message with a large timestamp.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "0".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0,
log.activeSegment.baseOffset))
+ // Append another message with a small timestamp and roll out a new log
segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() -
logConfig.deleteRetentionMs - 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1,
log.activeSegment.baseOffset))
+ assertEquals(0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ // Append a message to new key and roll out another log segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "1".getBytes,
+ timestamp = time.milliseconds()),
leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2,
log.activeSegment.baseOffset))
+ assertEquals(0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ // Append a message to key = 0 with lager timestamp and roll out another
log segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "2".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 20000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2,
log.activeSegment.baseOffset))
+ assertEquals(2,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ }
+
+ @Test
+ def testHeaderCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyHeader)
+ logProps.put(LogConfig.CompactionStrategyHeaderKeyProp, "sequence")
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a message with a large timestamp.
+ var headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(2L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "0".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0,
log.activeSegment.baseOffset))
+ // Append another message with a small timestamp and roll out a new log
segment.
+ headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(1L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = "0".getBytes,
+ key = "0".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds() -
logConfig.deleteRetentionMs - 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1,
log.activeSegment.baseOffset))
+ assertEquals(0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
Review comment:
Also, it seems that we should assert the message with the smaller header
is now removed?
##########
File path: core/src/test/scala/unit/kafka/log/FakeRecord.scala
##########
@@ -0,0 +1,74 @@
+/**
+ * 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 kafka.log
+
+import java.lang.{Long => JLong}
+import java.nio.ByteBuffer
+
+import kafka.utils.TestUtils
+import org.apache.kafka.common.header.Header
+import org.apache.kafka.common.header.internals.RecordHeader
+import org.apache.kafka.common.record.{Record, TimestampType}
+
+class FakeRecord(fakeKey: ByteBuffer, fakeOffset: Number, fakeSequence: Long =
-1L, fakeTimestamp: Long = -1L) extends Record {
+ private var fakeHeaders: Array[Header] = _
Review comment:
Instead of having a private init(), we could just do `fakeHeaders = if
() a else b`.
##########
File path: core/src/test/scala/unit/kafka/log/OffsetMapTest.scala
##########
@@ -28,54 +28,315 @@ class OffsetMapTest {
@Test
def testBasicValidation(): Unit = {
validateMap(10)
+ validateMap(10, strategy = "header", headerKey = "sequence")
+ validateMap(10, strategy = "timestamp")
validateMap(100)
+ validateMap(100, strategy = "header", headerKey = "sequence")
+ validateMap(100, strategy = "timestamp")
validateMap(1000)
+ validateMap(1000, strategy = "header", headerKey = "sequence")
+ validateMap(1000, strategy = "timestamp")
validateMap(5000)
+ validateMap(5000, strategy = "header", headerKey = "sequence")
+ validateMap(5000, strategy = "timestamp")
}
@Test
- def testClear(): Unit = {
+ def testInit(): Unit = {
val map = new SkimpyOffsetMap(4000)
- for(i <- 0 until 10)
- map.put(key(i), i)
- for(i <- 0 until 10)
- assertEquals(i.toLong, map.get(key(i)))
- map.clear()
- for(i <- 0 until 10)
- assertEquals(map.get(key(i)), -1L)
+ map.init()
+
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i))
+
+ for (i <- 0 until 10) {
+ assertEquals(i.toLong, map.getOffset(key(i)))
+ assertEquals(-1L, map.getVersion(key(i)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(i), i)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(i), i-1)))
+ }
+
+ map.init(Defaults.CompactionStrategyOffset)
+ for (i <- 0 until 10) {
+ assertEquals(map.getOffset(key(i)), -1L)
+ assertEquals(map.getVersion(key(i)), -1L)
+ }
}
@Test
def testGetWhenFull(): Unit = {
val map = new SkimpyOffsetMap(4096)
+ map.init(Defaults.CompactionStrategyOffset)
var i = 37L //any value would do
while (map.size < map.slots) {
- map.put(key(i), i)
+ map.put(new FakeRecord(key(i), i))
i = i + 1L
}
- assertEquals(map.get(key(i)), -1L)
- assertEquals(map.get(key(i-1L)), i-1L)
+
+ assertEquals(map.getOffset(key(i)), -1L)
+ assertEquals(map.getVersion(key(i)), -1L)
+ assertEquals(map.getOffset(key(i-1L)), i-1L)
+ assertEquals(map.getVersion(key(i-1L)), -1L)
+ assertNotEquals(map.latestOffset, i)
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(i-1L), i-1L)))
+ }
+
+ @Test
+ def testTimestampLogCompaction(): Unit = {
+ val map = new SkimpyOffsetMap(4096)
+ map.init("timestamp")
+
+ // put offset keys from 0 - 9 with corresponding offset 0-9 respectively
+ val time = System.currentTimeMillis()
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i, fakeTimestamp = time+i))
+
+ // map should hold offset 0 for key 0
+ assertEquals(0, map.getOffset(key(0)))
+ assertTrue(map.getVersion(key(0)) > 0)
+ // latest offset in map should be 9
+ assertEquals(9, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ // putting latest value for key=0 first followed by old value
+ val fr1 = new FakeRecord(key(0), 10, fakeTimestamp = time+10)
+ map.put(fr1)
+ val fr2 = new FakeRecord(key(0), 11, fakeTimestamp = time-1000)
+ map.put(fr2)
+
+ // map should hold offset 10 for key 0
+ assertEquals(10, map.getOffset(key(0)))
+ assertEquals(fr1.timestamp, map.getVersion(key(0)))
+ assertEquals(true, map.shouldRetainRecord(fr1))
+ assertEquals(false, map.shouldRetainRecord(fr2))
+
+ // latest offset in map should be 11
+ assertEquals(11, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ map.put(new FakeRecord(key(10), 12, fakeTimestamp = time+12))
+
+ // latest offset in map should be 12
+ assertEquals(12, map.latestOffset)
+ // size should 11
+ assertEquals(11, map.size)
+ }
+
+ @Test
+ def testSequenceLogCompaction(): Unit = {
+ val map = new SkimpyOffsetMap(4096)
+ map.init("header", "sequence")
+
+ // put offset keys from 0 - 9 with corresponding offset 0-9 respectively
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i, 100+i))
+
+ // putting latest value for key=0 first followed by old value
+ map.put(new FakeRecord(key(0), 10, 111))
+ map.put(new FakeRecord(key(0), 11, 110))
+
+ // map should hold offset 10 for key 0
+ assertEquals(10, map.getOffset(key(0)))
+ assertEquals(111, map.getVersion(key(0)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(0), 10, 111)))
+ // latest offset in map should be 11
+ assertEquals(11, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ // putting latest value for key=1 first followed by old value
+ map.put(new FakeRecord(key(1), 12, 112))
+ map.put(new FakeRecord(key(1), 13, 114))
+ map.put(new FakeRecord(key(1), 14, 115))
+ map.put(new FakeRecord(key(1), 15, 113))
+ map.put(new FakeRecord(key(1), 16, 120))
+ map.put(new FakeRecord(key(1), 17, 119))
+
+ // map should hold offset 16 for key 1
+ assertEquals(16, map.getOffset(key(1)))
+ assertEquals(120, map.getVersion(key(1)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(1), 16, 120)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(1), 17,
119)))
+ // latest offset in map should be 17
+ assertEquals(17, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ map.put(new FakeRecord(key(1), 18, 118))
+ map.put(new FakeRecord(key(1), 19, 121))
+ map.put(new FakeRecord(key(1), 20, 116))
+ map.put(new FakeRecord(key(1), 21, 117))
+
+ // map should hold offset 19 for key 1
+ assertEquals(19, map.getOffset(key(1)))
+ assertEquals(121, map.getVersion(key(1)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(1), 19, 121)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(1), 21,
117)))
+ // latest offset in map should be 21
+ assertEquals(21, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ map.put(new FakeRecord(key(10), 22, 22))
+
+ // latest offset in map should be 22
+ assertEquals(22, map.latestOffset)
+ // size should 11
+ assertEquals(11, map.size)
+
+ // checking negative header value
+ map.put(new FakeRecord(key(1), 23, -117))
+
+ // latest offset in map should be 23
+ assertEquals(23, map.latestOffset)
+ // map should hold offset 19 for key 1
+ assertEquals(19, map.getOffset(key(1)))
+ assertEquals(121, map.getVersion(key(1)))
+ // size should 11
+ assertEquals(11, map.size)
+ }
+
+ @Test
+ def testSameOffsetMapInstanceForDifferentCompactionStrategy(): Unit = {
+ val map = new SkimpyOffsetMap(4096)
+
+ /*
+ * offset based compaction
+ */
+ map.init()
+
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i))
+
+ // map should hold offset 0 for key 0
+ assertEquals(0, map.getOffset(key(0)))
+ assertEquals(-1L, map.getVersion(key(0)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(0), 0)))
+ // latest offset in map should be 9
+ assertEquals(9, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ // putting latest value for key=0 first followed by old value
+ map.put(new FakeRecord(key(0), 10))
+ map.put(new FakeRecord(key(0), 11))
+
+ // map should hold offset 11 for key 0
+ assertEquals(11, map.getOffset(key(0)))
+ assertEquals(-1L, map.getVersion(key(0)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(0), 10)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(0), 11)))
+ // latest offset in map should be 11
+ assertEquals(11, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ /*
+ * timestamp based compaction
+ */
+ map.init("timestamp")
+
+ // put offset keys from 0 - 9 with corresponding offset 0-9 respectively
+ val time = System.currentTimeMillis()
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i, fakeTimestamp = time+i))
+
+ // map should hold offset 0 for key 0
+ assertEquals(0, map.getOffset(key(0)))
+ assertTrue(map.getVersion(key(0)) > 0)
+ // latest offset in map should be 9
+ assertEquals(9, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ // putting latest value for key=0 first followed by old value
+ val fr1 = new FakeRecord(key(0), 10, fakeTimestamp = time+10)
+ map.put(fr1)
+ val fr2 = new FakeRecord(key(0), 11, fakeTimestamp = time-1000)
+ map.put(fr2)
+
+ // map should hold offset 10 for key 0
+ assertEquals(10, map.getOffset(key(0)))
+ assertEquals(fr1.timestamp, map.getVersion(key(0)))
+ assertEquals(true, map.shouldRetainRecord(fr1))
+ assertEquals(false, map.shouldRetainRecord(fr2))
+ // latest offset in map should be 11
+ assertEquals(11, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ /*
+ * header sequence based compaction
+ */
+ map.init("header", "sequence")
+
+ // put offset keys from 0 - 9 with corresponding offset 0-9 respectively
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i, 100+i))
+
+ // putting latest value for key=0 first followed by old value
+ map.put(new FakeRecord(key(0), 10, 111))
+ map.put(new FakeRecord(key(0), 11, 110))
+
+ // map should hold offset 10 for key 0
+ assertEquals(10, map.getOffset(key(0)))
+ assertEquals(111, map.getVersion(key(0)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(0), 10, 111)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(0), 11,
110)))
+ // latest offset in map should be 11
+ assertEquals(11, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
}
def key(key: Long) = ByteBuffer.wrap(key.toString.getBytes)
- def validateMap(items: Int, loadFactor: Double = 0.5): SkimpyOffsetMap = {
+ def validateMap(items: Int, loadFactor: Double = 0.5, strategy: String =
Defaults.CompactionStrategyOffset, headerKey: String = ""): SkimpyOffsetMap = {
val map = new SkimpyOffsetMap((items/loadFactor * 24).toInt)
- for(i <- 0 until items)
- map.put(key(i), i)
- for(i <- 0 until items)
- assertEquals(map.get(key(i)), i.toLong)
+ map.init(strategy, headerKey)
+
+ val isOffsetStrategy =
Defaults.CompactionStrategyOffset.equalsIgnoreCase(strategy)
+ val isTimestampStrategy =
Defaults.CompactionStrategyTimestamp.equalsIgnoreCase(strategy)
+ val currentTimeMillis = System.currentTimeMillis()
+ for (i <- 1 until items) {
+ if (isOffsetStrategy)
+ map.put(new FakeRecord(key(i), i))
+ else if (isTimestampStrategy)
+ map.put(new FakeRecord(key(i), i, fakeTimestamp = (currentTimeMillis +
i)))
+ else
+ map.put(new FakeRecord(key(i), i, items+i))
Review comment:
It's clearer to explicitly name the 3rd param. Also, `items+i `=> `items
+ i`. Ditto in a few other places.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
##########
@@ -848,6 +850,15 @@ class KafkaConfigTest {
assertFalse(isValidKafkaConfig(props))
}
+ @Test
+ def testInvalidCompactionHeaderKeyConfig(): Unit = {
Review comment:
testInvalidCompactionHeaderKeyConfig =>
testMissingCompactionHeaderKeyConfig ?
##########
File path: core/src/test/scala/unit/kafka/log/OffsetMapTest.scala
##########
@@ -28,54 +28,315 @@ class OffsetMapTest {
@Test
def testBasicValidation(): Unit = {
validateMap(10)
+ validateMap(10, strategy = "header", headerKey = "sequence")
+ validateMap(10, strategy = "timestamp")
validateMap(100)
+ validateMap(100, strategy = "header", headerKey = "sequence")
+ validateMap(100, strategy = "timestamp")
validateMap(1000)
+ validateMap(1000, strategy = "header", headerKey = "sequence")
+ validateMap(1000, strategy = "timestamp")
validateMap(5000)
+ validateMap(5000, strategy = "header", headerKey = "sequence")
+ validateMap(5000, strategy = "timestamp")
}
@Test
- def testClear(): Unit = {
+ def testInit(): Unit = {
val map = new SkimpyOffsetMap(4000)
- for(i <- 0 until 10)
- map.put(key(i), i)
- for(i <- 0 until 10)
- assertEquals(i.toLong, map.get(key(i)))
- map.clear()
- for(i <- 0 until 10)
- assertEquals(map.get(key(i)), -1L)
+ map.init()
+
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i))
+
+ for (i <- 0 until 10) {
+ assertEquals(i.toLong, map.getOffset(key(i)))
+ assertEquals(-1L, map.getVersion(key(i)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(i), i)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(i), i-1)))
+ }
+
+ map.init(Defaults.CompactionStrategyOffset)
+ for (i <- 0 until 10) {
+ assertEquals(map.getOffset(key(i)), -1L)
+ assertEquals(map.getVersion(key(i)), -1L)
+ }
}
@Test
def testGetWhenFull(): Unit = {
val map = new SkimpyOffsetMap(4096)
+ map.init(Defaults.CompactionStrategyOffset)
var i = 37L //any value would do
while (map.size < map.slots) {
- map.put(key(i), i)
+ map.put(new FakeRecord(key(i), i))
i = i + 1L
}
- assertEquals(map.get(key(i)), -1L)
- assertEquals(map.get(key(i-1L)), i-1L)
+
+ assertEquals(map.getOffset(key(i)), -1L)
+ assertEquals(map.getVersion(key(i)), -1L)
+ assertEquals(map.getOffset(key(i-1L)), i-1L)
+ assertEquals(map.getVersion(key(i-1L)), -1L)
+ assertNotEquals(map.latestOffset, i)
Review comment:
It seems it's better to assert what the latestOffset is equal to.
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -1560,6 +2040,160 @@ class LogCleanerTest {
assertEquals("The tombstone should be retained.", 1,
log.logSegments.head.log.batches.iterator.next().lastOffset)
}
+ @Test
+ def testCleanTombstoneWithTimestampCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyTimestamp)
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a tombstone with a large timestamp.
+ log.appendAsLeader(TestUtils.singletonRecords(value = null,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0,
log.activeSegment.baseOffset))
+ // Append a message with a small timestamp and roll out a new log segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "0".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() -
logConfig.deleteRetentionMs - 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ // Append a message and roll out another log segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "1".getBytes,
+ timestamp = time.milliseconds()),
leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ }
+
+ @Test
+ def testCleanTombstoneWithHeaderCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyHeader)
+ logProps.put(LogConfig.CompactionStrategyHeaderKeyProp, "sequence")
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a message with a large timestamp.
+ var headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(1L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = "0".getBytes,
+ key = "0".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0,
log.activeSegment.baseOffset))
+ // Append a tombstone with a small timestamp and roll out a new log
segment.
+ headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(2L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = null,
+ key = "0".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds() -
logConfig.deleteRetentionMs - 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 1,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ // Append a message and roll out another log segment.
+ headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(3L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "1".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds()),
leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 1,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ }
+
+ @Test
+ def testTimestampCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyTimestamp)
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a message with a large timestamp.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "0".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0,
log.activeSegment.baseOffset))
+ // Append another message with a small timestamp and roll out a new log
segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() -
logConfig.deleteRetentionMs - 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1,
log.activeSegment.baseOffset))
+ assertEquals(0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
Review comment:
Could we add an error message describing what's expected? Ditto in a few
other places. Also, it seems that we should assert the message with the smaller
timestamp is now removed?
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -162,9 +164,28 @@ class LogCleanerTest {
@Test
def testSizeTrimmedForPreallocatedAndCompactedTopic(): Unit = {
+ val logProps = new Properties()
+ testSizeTrimmedForPreallocatedAndCompactedTopic(logProps)
+ }
+
+ @Test
+ def
testSizeTrimmedForPreallocatedAndCompactedTopicWithTimestampCompaction(): Unit
= {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyTimestamp)
+ testSizeTrimmedForPreallocatedAndCompactedTopic(logProps)
Review comment:
Is this test useful? We don't set the timestamp in the test.
##########
File path: core/src/test/scala/unit/kafka/log/FakeRecord.scala
##########
@@ -0,0 +1,74 @@
+/**
+ * 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 kafka.log
+
+import java.lang.{Long => JLong}
+import java.nio.ByteBuffer
+
+import kafka.utils.TestUtils
+import org.apache.kafka.common.header.Header
+import org.apache.kafka.common.header.internals.RecordHeader
+import org.apache.kafka.common.record.{Record, TimestampType}
+
+class FakeRecord(fakeKey: ByteBuffer, fakeOffset: Number, fakeSequence: Long =
-1L, fakeTimestamp: Long = -1L) extends Record {
+ private var fakeHeaders: Array[Header] = _
+
+ init()
+
+ private def init(): Unit = {
+ if (Option(fakeSequence).isEmpty || fakeSequence <= 0)
+ fakeHeaders = new Array[Header](0)
+ else
+ fakeHeaders = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(fakeSequence)))
+ }
+
+ override def offset: Long = if (fakeOffset != null) fakeOffset.longValue
else -1
Review comment:
No need for `fakeOffset != null` since it's not expected to be null.
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -1725,26 +2360,64 @@ class LogCleanerTest {
}
class FakeOffsetMap(val slots: Int) extends OffsetMap {
- val map = new java.util.HashMap[String, Long]()
+ val map = new java.util.HashMap[String, Record]()
var lastOffset = -1L
+ var isOffsetStrategy = true
+ var isTimestampStrategy = false
+ var headerKey = ""
private def keyFor(key: ByteBuffer) =
new String(Utils.readBytes(key.duplicate), StandardCharsets.UTF_8)
- override def put(key: ByteBuffer, offset: Long): Unit = {
- lastOffset = offset
- map.put(keyFor(key), offset)
+ override def init(strategy: String = Defaults.CompactionStrategyOffset,
headerKey: String = "", cleanerThreadId: Int = -1, topicPartitionName: String =
"") = {
+ this.map.clear()
+
+ this.isOffsetStrategy =
Defaults.CompactionStrategyOffset.equalsIgnoreCase(strategy)
+ this.isTimestampStrategy =
Defaults.CompactionStrategyTimestamp.equalsIgnoreCase(strategy)
+ this.headerKey = headerKey
}
- override def get(key: ByteBuffer): Long = {
+ override def put(record: Record): Boolean = {
+ lastOffset = record.offset
+ extractVersion(record)
+ val key = keyFor(record.key)
+ if (!isOffsetStrategy && map.containsKey(key)) {
+ val foundVersion = extractVersion(map.get(key))
+ val currVersion = extractVersion(record)
+ if (foundVersion > currVersion)
Review comment:
This needs to be adjusted to take the larger offset when versions equal.
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -1560,6 +2040,160 @@ class LogCleanerTest {
assertEquals("The tombstone should be retained.", 1,
log.logSegments.head.log.batches.iterator.next().lastOffset)
}
+ @Test
+ def testCleanTombstoneWithTimestampCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyTimestamp)
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a tombstone with a large timestamp.
+ log.appendAsLeader(TestUtils.singletonRecords(value = null,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0,
log.activeSegment.baseOffset))
+ // Append a message with a small timestamp and roll out a new log segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "0".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() -
logConfig.deleteRetentionMs - 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ // Append a message and roll out another log segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "1".getBytes,
+ timestamp = time.milliseconds()),
leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ }
+
+ @Test
+ def testCleanTombstoneWithHeaderCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyHeader)
+ logProps.put(LogConfig.CompactionStrategyHeaderKeyProp, "sequence")
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a message with a large timestamp.
+ var headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(1L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = "0".getBytes,
+ key = "0".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0,
log.activeSegment.baseOffset))
+ // Append a tombstone with a small timestamp and roll out a new log
segment.
+ headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(2L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = null,
+ key = "0".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds() -
logConfig.deleteRetentionMs - 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 1,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ // Append a message and roll out another log segment.
+ headers = Array[Header](new RecordHeader("sequence",
TestUtils.longToByte(3L)))
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "1".getBytes,
+ headers = headers,
+ timestamp = time.milliseconds()),
leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2,
log.activeSegment.baseOffset))
+ assertEquals("The tombstone should be retained.", 1,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ }
+
+ @Test
+ def testTimestampCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyTimestamp)
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a message with a large timestamp.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "0".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0,
log.activeSegment.baseOffset))
+ // Append another message with a small timestamp and roll out a new log
segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() -
logConfig.deleteRetentionMs - 10000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1,
log.activeSegment.baseOffset))
+ assertEquals(0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ // Append a message to new key and roll out another log segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
+ key = "1".getBytes,
+ timestamp = time.milliseconds()),
leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2,
log.activeSegment.baseOffset))
+ assertEquals(0,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ // Append a message to key = 0 with lager timestamp and roll out another
log segment.
+ log.appendAsLeader(TestUtils.singletonRecords(value = "2".getBytes,
+ key = "0".getBytes,
+ timestamp = time.milliseconds() +
logConfig.deleteRetentionMs + 20000), leaderEpoch = 0)
+ log.roll()
+ cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2,
log.activeSegment.baseOffset))
+ assertEquals(2,
log.logSegments.head.log.batches.iterator.next().lastOffset)
+ }
+
+ @Test
+ def testHeaderCompaction(): Unit = {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyHeader)
+ logProps.put(LogConfig.CompactionStrategyHeaderKeyProp, "sequence")
+ val logConfig = LogConfig(logProps)
+
+ val log = makeLog(config = logConfig)
+ val cleaner = makeCleaner(10)
+
+ // Append a message with a large timestamp.
Review comment:
inaccurate comment.
##########
File path: core/src/main/scala/kafka/log/OffsetMap.scala
##########
@@ -68,90 +120,164 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm:
String = "MD5") extend
/**
* The number of bytes of space each entry uses (the number of bytes in the
hash plus an 8 byte offset)
+ * This evaluates to the number of bytes in the hash plus 8 bytes for the
offset
+ * and, if applicable, another 8 bytes for non-offset compact strategy (set
in the init method).
*/
- val bytesPerEntry = hashSize + 8
-
+ var bytesPerEntry = hashSize + longByteSize
+
/**
* The maximum number of entries this map can contain
*/
- val slots: Int = memory / bytesPerEntry
+ var slots: Int = memory / bytesPerEntry
+
+ /* compact strategy */
+ private var compactionStrategy: CompactionStrategy = null
+
+ /* header key for the Strategy header to look for */
+ private var headerKey: String = ""
+
+ /**
+ * Initialize the map with the topic compact strategy
+ * @param strategy The compaction strategy
+ * @param headerKey The header key if the compaction strategy is set to
header
+ * @param cleanerThreadId The clenaer thread id
+ * @param topicPartitionName The topic partition name
+ */
+ override def init(strategy: String = Defaults.CompactionStrategy, headerKey:
String = "", cleanerThreadId: Int = -1, topicPartitionName: String = "") {
+ // set the log indent for the topic partition
+ this.logIdent = s"[OffsetMap-$cleanerThreadId $topicPartitionName]: "
+ info(s"Initializing OffsetMap with compaction strategy '$strategy' and
header key '$headerKey'")
+
+ // Change the salt used for key hashing making all existing keys
unfindable.
+ this.entries = 0
+ this.lookups = 0L
+ this.probes = 0L
+ this.lastOffset = -1L
+ Arrays.fill(bytes.array, bytes.arrayOffset, bytes.arrayOffset +
bytes.limit(), 0.toByte)
+
+ this.compactionStrategy = CompactionStrategy.withName(strategy)
+ this.headerKey = headerKey.trim()
+
+ info(s"Compaction strategy set to '${this.compactionStrategy}'")
+ this.bytesPerEntry = hashSize + longByteSize + (if
(this.compactionStrategy == CompactionStrategy.OFFSET) 0 else longByteSize)
+ this.slots = memory / bytesPerEntry
+ }
/**
* Associate this offset to the given key.
- * @param key The key
- * @param offset The offset
+ * @param record The record
+ * @return success flag
*/
- override def put(key: ByteBuffer, offset: Long): Unit = {
+ override def put(record: Record): Boolean = {
require(entries < slots, "Attempt to add a new entry to a full offset
map.")
+
+ val key = record.key
+ val offset = record.offset
+ val currVersion = extractVersion(record)
+
+ lastOffset = offset
lookups += 1
hashInto(key, hash1)
+
// probe until we find the first empty slot
var attempt = 0
var pos = positionOf(hash1, attempt)
- while(!isEmpty(pos)) {
+ while (!isEmpty(pos)) {
bytes.position(pos)
bytes.get(hash2)
- if(Arrays.equals(hash1, hash2)) {
+ if (Arrays.equals(hash1, hash2)) {
+ // we found an existing entry, overwrite it and return (size does not
change)
+ if (this.compactionStrategy != CompactionStrategy.OFFSET) {
+ // read previous value by skipping offset
+ bytes.position(bytes.position() + longByteSize)
+ val foundVersion = bytes.getLong()
+ if (foundVersion > currVersion) {
+ // map already holding latest record
+ return false
+ }
+
+ // reset the position to start of offset
+ bytes.position(bytes.position() - (2*longByteSize))
+ }
+
// we found an existing entry, overwrite it and return (size does not
change)
bytes.putLong(offset)
- lastOffset = offset
- return
+ if (this.compactionStrategy != CompactionStrategy.OFFSET) {
+ bytes.putLong(currVersion)
+ }
+
+ return true
}
+
attempt += 1
pos = positionOf(hash1, attempt)
}
+
// found an empty slot, update it--size grows by 1
bytes.position(pos)
bytes.put(hash1)
bytes.putLong(offset)
- lastOffset = offset
+ if (this.compactionStrategy != CompactionStrategy.OFFSET) {
+ bytes.putLong(currVersion)
+ }
+
entries += 1
+ true
}
-
+
/**
* Check that there is no entry at the given position
*/
private def isEmpty(position: Int): Boolean =
- bytes.getLong(position) == 0 && bytes.getLong(position + 8) == 0 &&
bytes.getLong(position + 16) == 0
+ bytes.getLong(position) == 0 && bytes.getLong(position + longByteSize) ==
0 && bytes.getLong(position + (2*longByteSize)) == 0
/**
- * Get the offset associated with this key.
+ * Checks to see whether to retain the record or not
+ * @param record The record
+ * @return true to retain; false not to
+ */
+ override def shouldRetainRecord(record: Record): Boolean = {
+ if (this.compactionStrategy != CompactionStrategy.OFFSET) {
+ val foundVersion = getVersion(record.key)
+ val currentVersion = extractVersion(record)
+ // use version if available & different otherwise fallback to offset
Review comment:
The case I am thinking is the following. Suppose in log segment 1, we
have (offset, key, timestamp) (10, 1, 100). We do a round of timestamp
compaction and keep this record. In log segment 2, we have (20, 1, 99). When
building the map, we only load the timestamp 99 for key 1. This means that
during cleaning, offset 20 won't be removed, but it seems that it should be.
Also, according to the KIP, we need to handle LEO specially. If the record for
LEO is going to be removed, we need to (1) retain an empty batch to keep LEO if
the message format is 2 or above; (2) keep the message if the format is below 2.
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -162,9 +164,28 @@ class LogCleanerTest {
@Test
def testSizeTrimmedForPreallocatedAndCompactedTopic(): Unit = {
+ val logProps = new Properties()
+ testSizeTrimmedForPreallocatedAndCompactedTopic(logProps)
+ }
+
+ @Test
+ def
testSizeTrimmedForPreallocatedAndCompactedTopicWithTimestampCompaction(): Unit
= {
+ val logProps = new Properties()
+ logProps.put(LogConfig.CompactionStrategyProp,
Defaults.CompactionStrategyTimestamp)
+ testSizeTrimmedForPreallocatedAndCompactedTopic(logProps)
+ }
+
+ @Test
+ def testSizeTrimmedForPreallocatedAndCompactedTopicWithHeaderCompaction():
Unit = {
Review comment:
Is this test useful? We don't set the header value in the test.
##########
File path: core/src/test/scala/unit/kafka/log/LogConfigTest.scala
##########
@@ -93,6 +95,15 @@ class LogConfigTest {
}
}
+ @Test
+ def testInvalidCompactionHeaderKeyConfig(): Unit = {
+ val props = new Properties
+ props.setProperty(LogConfig.CompactionStrategyProp, "header")
+ intercept[Exception] {
Review comment:
Could we intercept InvalidConfigException?
##########
File path: core/src/test/scala/unit/kafka/log/OffsetMapTest.scala
##########
@@ -28,54 +28,315 @@ class OffsetMapTest {
@Test
def testBasicValidation(): Unit = {
validateMap(10)
+ validateMap(10, strategy = "header", headerKey = "sequence")
+ validateMap(10, strategy = "timestamp")
validateMap(100)
+ validateMap(100, strategy = "header", headerKey = "sequence")
+ validateMap(100, strategy = "timestamp")
validateMap(1000)
+ validateMap(1000, strategy = "header", headerKey = "sequence")
+ validateMap(1000, strategy = "timestamp")
validateMap(5000)
+ validateMap(5000, strategy = "header", headerKey = "sequence")
+ validateMap(5000, strategy = "timestamp")
}
@Test
- def testClear(): Unit = {
+ def testInit(): Unit = {
val map = new SkimpyOffsetMap(4000)
- for(i <- 0 until 10)
- map.put(key(i), i)
- for(i <- 0 until 10)
- assertEquals(i.toLong, map.get(key(i)))
- map.clear()
- for(i <- 0 until 10)
- assertEquals(map.get(key(i)), -1L)
+ map.init()
+
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i))
+
+ for (i <- 0 until 10) {
+ assertEquals(i.toLong, map.getOffset(key(i)))
+ assertEquals(-1L, map.getVersion(key(i)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(i), i)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(i), i-1)))
+ }
+
+ map.init(Defaults.CompactionStrategyOffset)
+ for (i <- 0 until 10) {
+ assertEquals(map.getOffset(key(i)), -1L)
+ assertEquals(map.getVersion(key(i)), -1L)
+ }
}
@Test
def testGetWhenFull(): Unit = {
val map = new SkimpyOffsetMap(4096)
+ map.init(Defaults.CompactionStrategyOffset)
var i = 37L //any value would do
while (map.size < map.slots) {
- map.put(key(i), i)
+ map.put(new FakeRecord(key(i), i))
i = i + 1L
}
- assertEquals(map.get(key(i)), -1L)
- assertEquals(map.get(key(i-1L)), i-1L)
+
+ assertEquals(map.getOffset(key(i)), -1L)
+ assertEquals(map.getVersion(key(i)), -1L)
+ assertEquals(map.getOffset(key(i-1L)), i-1L)
+ assertEquals(map.getVersion(key(i-1L)), -1L)
+ assertNotEquals(map.latestOffset, i)
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(i-1L), i-1L)))
+ }
+
+ @Test
+ def testTimestampLogCompaction(): Unit = {
+ val map = new SkimpyOffsetMap(4096)
+ map.init("timestamp")
+
+ // put offset keys from 0 - 9 with corresponding offset 0-9 respectively
+ val time = System.currentTimeMillis()
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i, fakeTimestamp = time+i))
+
+ // map should hold offset 0 for key 0
+ assertEquals(0, map.getOffset(key(0)))
+ assertTrue(map.getVersion(key(0)) > 0)
+ // latest offset in map should be 9
+ assertEquals(9, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ // putting latest value for key=0 first followed by old value
+ val fr1 = new FakeRecord(key(0), 10, fakeTimestamp = time+10)
+ map.put(fr1)
+ val fr2 = new FakeRecord(key(0), 11, fakeTimestamp = time-1000)
+ map.put(fr2)
+
+ // map should hold offset 10 for key 0
+ assertEquals(10, map.getOffset(key(0)))
+ assertEquals(fr1.timestamp, map.getVersion(key(0)))
+ assertEquals(true, map.shouldRetainRecord(fr1))
+ assertEquals(false, map.shouldRetainRecord(fr2))
+
+ // latest offset in map should be 11
+ assertEquals(11, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ map.put(new FakeRecord(key(10), 12, fakeTimestamp = time+12))
+
+ // latest offset in map should be 12
+ assertEquals(12, map.latestOffset)
+ // size should 11
+ assertEquals(11, map.size)
+ }
+
+ @Test
+ def testSequenceLogCompaction(): Unit = {
+ val map = new SkimpyOffsetMap(4096)
+ map.init("header", "sequence")
+
+ // put offset keys from 0 - 9 with corresponding offset 0-9 respectively
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i, 100+i))
+
+ // putting latest value for key=0 first followed by old value
+ map.put(new FakeRecord(key(0), 10, 111))
+ map.put(new FakeRecord(key(0), 11, 110))
+
+ // map should hold offset 10 for key 0
+ assertEquals(10, map.getOffset(key(0)))
+ assertEquals(111, map.getVersion(key(0)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(0), 10, 111)))
+ // latest offset in map should be 11
+ assertEquals(11, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ // putting latest value for key=1 first followed by old value
Review comment:
The first value doesn't seem to be the latest.
##########
File path: core/src/test/scala/unit/kafka/log/OffsetMapTest.scala
##########
@@ -28,54 +28,315 @@ class OffsetMapTest {
@Test
def testBasicValidation(): Unit = {
validateMap(10)
+ validateMap(10, strategy = "header", headerKey = "sequence")
+ validateMap(10, strategy = "timestamp")
validateMap(100)
+ validateMap(100, strategy = "header", headerKey = "sequence")
+ validateMap(100, strategy = "timestamp")
validateMap(1000)
+ validateMap(1000, strategy = "header", headerKey = "sequence")
+ validateMap(1000, strategy = "timestamp")
validateMap(5000)
+ validateMap(5000, strategy = "header", headerKey = "sequence")
+ validateMap(5000, strategy = "timestamp")
}
@Test
- def testClear(): Unit = {
+ def testInit(): Unit = {
val map = new SkimpyOffsetMap(4000)
- for(i <- 0 until 10)
- map.put(key(i), i)
- for(i <- 0 until 10)
- assertEquals(i.toLong, map.get(key(i)))
- map.clear()
- for(i <- 0 until 10)
- assertEquals(map.get(key(i)), -1L)
+ map.init()
+
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i))
+
+ for (i <- 0 until 10) {
+ assertEquals(i.toLong, map.getOffset(key(i)))
+ assertEquals(-1L, map.getVersion(key(i)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(i), i)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(i), i-1)))
+ }
+
+ map.init(Defaults.CompactionStrategyOffset)
+ for (i <- 0 until 10) {
+ assertEquals(map.getOffset(key(i)), -1L)
+ assertEquals(map.getVersion(key(i)), -1L)
+ }
}
@Test
def testGetWhenFull(): Unit = {
val map = new SkimpyOffsetMap(4096)
+ map.init(Defaults.CompactionStrategyOffset)
var i = 37L //any value would do
while (map.size < map.slots) {
- map.put(key(i), i)
+ map.put(new FakeRecord(key(i), i))
i = i + 1L
}
- assertEquals(map.get(key(i)), -1L)
- assertEquals(map.get(key(i-1L)), i-1L)
+
+ assertEquals(map.getOffset(key(i)), -1L)
+ assertEquals(map.getVersion(key(i)), -1L)
+ assertEquals(map.getOffset(key(i-1L)), i-1L)
+ assertEquals(map.getVersion(key(i-1L)), -1L)
+ assertNotEquals(map.latestOffset, i)
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(i-1L), i-1L)))
+ }
+
+ @Test
+ def testTimestampLogCompaction(): Unit = {
+ val map = new SkimpyOffsetMap(4096)
+ map.init("timestamp")
+
+ // put offset keys from 0 - 9 with corresponding offset 0-9 respectively
+ val time = System.currentTimeMillis()
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i, fakeTimestamp = time+i))
+
+ // map should hold offset 0 for key 0
+ assertEquals(0, map.getOffset(key(0)))
+ assertTrue(map.getVersion(key(0)) > 0)
+ // latest offset in map should be 9
+ assertEquals(9, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ // putting latest value for key=0 first followed by old value
+ val fr1 = new FakeRecord(key(0), 10, fakeTimestamp = time+10)
+ map.put(fr1)
+ val fr2 = new FakeRecord(key(0), 11, fakeTimestamp = time-1000)
+ map.put(fr2)
+
+ // map should hold offset 10 for key 0
+ assertEquals(10, map.getOffset(key(0)))
+ assertEquals(fr1.timestamp, map.getVersion(key(0)))
+ assertEquals(true, map.shouldRetainRecord(fr1))
+ assertEquals(false, map.shouldRetainRecord(fr2))
+
+ // latest offset in map should be 11
+ assertEquals(11, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ map.put(new FakeRecord(key(10), 12, fakeTimestamp = time+12))
+
+ // latest offset in map should be 12
+ assertEquals(12, map.latestOffset)
+ // size should 11
+ assertEquals(11, map.size)
+ }
+
+ @Test
+ def testSequenceLogCompaction(): Unit = {
+ val map = new SkimpyOffsetMap(4096)
+ map.init("header", "sequence")
+
+ // put offset keys from 0 - 9 with corresponding offset 0-9 respectively
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i, 100+i))
+
+ // putting latest value for key=0 first followed by old value
+ map.put(new FakeRecord(key(0), 10, 111))
+ map.put(new FakeRecord(key(0), 11, 110))
+
+ // map should hold offset 10 for key 0
+ assertEquals(10, map.getOffset(key(0)))
+ assertEquals(111, map.getVersion(key(0)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(0), 10, 111)))
+ // latest offset in map should be 11
+ assertEquals(11, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ // putting latest value for key=1 first followed by old value
+ map.put(new FakeRecord(key(1), 12, 112))
+ map.put(new FakeRecord(key(1), 13, 114))
+ map.put(new FakeRecord(key(1), 14, 115))
+ map.put(new FakeRecord(key(1), 15, 113))
+ map.put(new FakeRecord(key(1), 16, 120))
+ map.put(new FakeRecord(key(1), 17, 119))
+
+ // map should hold offset 16 for key 1
+ assertEquals(16, map.getOffset(key(1)))
+ assertEquals(120, map.getVersion(key(1)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(1), 16, 120)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(1), 17,
119)))
+ // latest offset in map should be 17
+ assertEquals(17, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ map.put(new FakeRecord(key(1), 18, 118))
+ map.put(new FakeRecord(key(1), 19, 121))
+ map.put(new FakeRecord(key(1), 20, 116))
+ map.put(new FakeRecord(key(1), 21, 117))
+
+ // map should hold offset 19 for key 1
+ assertEquals(19, map.getOffset(key(1)))
+ assertEquals(121, map.getVersion(key(1)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(1), 19, 121)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(1), 21,
117)))
+ // latest offset in map should be 21
+ assertEquals(21, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ map.put(new FakeRecord(key(10), 22, 22))
+
+ // latest offset in map should be 22
+ assertEquals(22, map.latestOffset)
+ // size should 11
+ assertEquals(11, map.size)
+
+ // checking negative header value
+ map.put(new FakeRecord(key(1), 23, -117))
+
+ // latest offset in map should be 23
+ assertEquals(23, map.latestOffset)
+ // map should hold offset 19 for key 1
+ assertEquals(19, map.getOffset(key(1)))
+ assertEquals(121, map.getVersion(key(1)))
+ // size should 11
+ assertEquals(11, map.size)
+ }
+
+ @Test
+ def testSameOffsetMapInstanceForDifferentCompactionStrategy(): Unit = {
+ val map = new SkimpyOffsetMap(4096)
+
+ /*
+ * offset based compaction
+ */
+ map.init()
+
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i))
+
+ // map should hold offset 0 for key 0
+ assertEquals(0, map.getOffset(key(0)))
+ assertEquals(-1L, map.getVersion(key(0)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(0), 0)))
+ // latest offset in map should be 9
+ assertEquals(9, map.latestOffset)
+ // size should 10
+ assertEquals(10, map.size)
+
+ // putting latest value for key=0 first followed by old value
Review comment:
The comment seems inaccurate.
##########
File path: core/src/test/scala/unit/kafka/log/OffsetMapTest.scala
##########
@@ -28,54 +28,315 @@ class OffsetMapTest {
@Test
def testBasicValidation(): Unit = {
validateMap(10)
+ validateMap(10, strategy = "header", headerKey = "sequence")
+ validateMap(10, strategy = "timestamp")
validateMap(100)
+ validateMap(100, strategy = "header", headerKey = "sequence")
+ validateMap(100, strategy = "timestamp")
validateMap(1000)
+ validateMap(1000, strategy = "header", headerKey = "sequence")
+ validateMap(1000, strategy = "timestamp")
validateMap(5000)
+ validateMap(5000, strategy = "header", headerKey = "sequence")
+ validateMap(5000, strategy = "timestamp")
}
@Test
- def testClear(): Unit = {
+ def testInit(): Unit = {
val map = new SkimpyOffsetMap(4000)
- for(i <- 0 until 10)
- map.put(key(i), i)
- for(i <- 0 until 10)
- assertEquals(i.toLong, map.get(key(i)))
- map.clear()
- for(i <- 0 until 10)
- assertEquals(map.get(key(i)), -1L)
+ map.init()
+
+ for (i <- 0 until 10)
+ map.put(new FakeRecord(key(i), i))
+
+ for (i <- 0 until 10) {
+ assertEquals(i.toLong, map.getOffset(key(i)))
+ assertEquals(-1L, map.getVersion(key(i)))
+ assertEquals(true, map.shouldRetainRecord(new FakeRecord(key(i), i)))
+ assertEquals(false, map.shouldRetainRecord(new FakeRecord(key(i), i-1)))
+ }
+
+ map.init(Defaults.CompactionStrategyOffset)
+ for (i <- 0 until 10) {
+ assertEquals(map.getOffset(key(i)), -1L)
+ assertEquals(map.getVersion(key(i)), -1L)
Review comment:
The expected value should be the first param. Ditto in a few other
places.
##########
File path: core/src/test/scala/unit/kafka/log/FakeRecord.scala
##########
@@ -0,0 +1,74 @@
+/**
+ * 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 kafka.log
+
+import java.lang.{Long => JLong}
+import java.nio.ByteBuffer
+
+import kafka.utils.TestUtils
+import org.apache.kafka.common.header.Header
+import org.apache.kafka.common.header.internals.RecordHeader
+import org.apache.kafka.common.record.{Record, TimestampType}
+
+class FakeRecord(fakeKey: ByteBuffer, fakeOffset: Number, fakeSequence: Long =
-1L, fakeTimestamp: Long = -1L) extends Record {
Review comment:
fakeSequence => fakeHeader?
##########
File path: core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
##########
@@ -1755,4 +2428,21 @@ class FakeOffsetMap(val slots: Int) extends OffsetMap {
}
override def toString: String = map.toString
+
+ private def extractVersion(record: Record): Long = {
Review comment:
This seems to be identical to what's in SkimpyOffsetMap. Perhaps we can
extract this as a utility method in OffsetMap and reuse? Ditto for
shouldRetainRecord().
----------------------------------------------------------------
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:
[email protected]
