You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/12/30 00:56:39 UTC
[GitHub] [kafka] jsancio opened a new pull request, #13058: KAFKA-14557; Lock metadata log dir
jsancio opened a new pull request, #13058:
URL: https://github.com/apache/kafka/pull/13058
This change makes sure that Kafka grabs a log dir lock in the following additional cases:
1. When a Kafka node runs in controller only. The current implementation doesn't grab a file lock because the LogManager is never instantiated.
2. When the metadata log dir is different from the log dir(s). The current implementation of LogManager doesn't load or grab a lock on the metadata dir.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on pull request #13058: KAFKA-14557; Lock metadata log dir
Posted by GitBox <gi...@apache.org>.
jsancio commented on PR #13058:
URL: https://github.com/apache/kafka/pull/13058#issuecomment-1371500365
I should mention that the long term solution would be to extend the LogManager to support the kraft metadata log but there are issues like https://issues.apache.org/jira/browse/KAFKA-14241 that need to be fixed to make that possible.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dengziming commented on a diff in pull request #13058: KAFKA-14557; Lock metadata log dir
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #13058:
URL: https://github.com/apache/kafka/pull/13058#discussion_r1059224316
##########
core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala:
##########
@@ -81,25 +100,99 @@ class RaftManagerTest {
)
}
- @Test
- def testNodeIdPresentIfBrokerRoleOnly(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "broker", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ @ParameterizedTest
+ @ValueSource(strings = Array("broker", "controller", "broker,controller"))
+ def testNodeIdPresent(processRoles: String): Unit = {
+ var processRolesSet = Set.empty[ProcessRole]
+ if (processRoles.contains("broker")) {
+ processRolesSet = processRolesSet ++ Set(BrokerRole)
+ }
+ if (processRoles.contains("controller")) {
+ processRolesSet = processRolesSet ++ Set(ControllerRole)
+ }
+
+ val logDir = TestUtils.tempDir()
+ val nodeId = 1
+ val raftManager = createRaftManager(
+ new TopicPartition("__raft_id_test", 0),
+ createConfig(
+ processRolesSet,
+ nodeId,
+ Some(logDir.toPath),
+ None
+ )
+ )
+ assertEquals(nodeId, raftManager.client.nodeId.getAsInt)
raftManager.shutdown()
}
- @Test
- def testNodeIdPresentIfControllerRoleOnly(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "controller", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ @ParameterizedTest
+ @ValueSource(strings = Array("metadata", "log", "metadata,log"))
+ def testLogDirLockWhenControllerOnly(dirType: String): Unit = {
+ val logDir = if (dirType.contains("metadata")) {
+ Some(TestUtils.tempDir().toPath)
+ } else {
+ None
+ }
+
+ val metadataDir = if (dirType.contains("log")) {
+ Some(TestUtils.tempDir().toPath)
+ } else {
+ None
+ }
+
+ val nodeId = 1
+ val raftManager = createRaftManager(
+ new TopicPartition("__raft_id_test", 0),
+ createConfig(
+ Set(ControllerRole),
+ nodeId,
+ logDir,
+ metadataDir
+ )
+ )
+
+ val lockPath = metadataDir.getOrElse(logDir.get).resolve(LogManager.LockFileName)
+ assertTrue(fileLocked(lockPath))
+
raftManager.shutdown()
+
+ assertFalse(fileLocked(lockPath))
}
@Test
- def testNodeIdPresentIfColocated(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "controller,broker", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ def testLogDirLockWhenMetadataDir(): Unit = {
+ val logDir = Some(TestUtils.tempDir().toPath)
+ val metadataDir = Some(TestUtils.tempDir().toPath)
+
+ val nodeId = 1
+ val raftManager = createRaftManager(
+ new TopicPartition("__raft_id_test", 0),
+ createConfig(
+ Set(BrokerRole),
Review Comment:
We are testing controller error so should this be ControllerRole.
##########
core/src/main/scala/kafka/raft/RaftManager.scala:
##########
@@ -154,6 +192,8 @@ class KafkaRaftManager[T](
scheduler.shutdown()
netChannel.close()
replicatedLog.close()
+
+ dataDirLock.foreach(_.destroy())
Review Comment:
In `LogManager` we do this in a finally block, should we do it here?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #13058: KAFKA-14557; Lock metadata log dir
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #13058:
URL: https://github.com/apache/kafka/pull/13058#discussion_r1061930398
##########
core/src/main/scala/kafka/raft/RaftManager.scala:
##########
@@ -154,6 +192,8 @@ class KafkaRaftManager[T](
scheduler.shutdown()
netChannel.close()
replicatedLog.close()
+
+ dataDirLock.foreach(_.destroy())
Review Comment:
Okay. I'll add `CoreUtils.swallow` to all of this calls.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #13058: KAFKA-14557; Lock metadata log dir
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #13058:
URL: https://github.com/apache/kafka/pull/13058#discussion_r1061928077
##########
core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala:
##########
@@ -81,25 +100,99 @@ class RaftManagerTest {
)
}
- @Test
- def testNodeIdPresentIfBrokerRoleOnly(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "broker", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ @ParameterizedTest
+ @ValueSource(strings = Array("broker", "controller", "broker,controller"))
+ def testNodeIdPresent(processRoles: String): Unit = {
+ var processRolesSet = Set.empty[ProcessRole]
+ if (processRoles.contains("broker")) {
+ processRolesSet = processRolesSet ++ Set(BrokerRole)
+ }
+ if (processRoles.contains("controller")) {
+ processRolesSet = processRolesSet ++ Set(ControllerRole)
+ }
+
+ val logDir = TestUtils.tempDir()
+ val nodeId = 1
+ val raftManager = createRaftManager(
+ new TopicPartition("__raft_id_test", 0),
+ createConfig(
+ processRolesSet,
+ nodeId,
+ Some(logDir.toPath),
+ None
+ )
+ )
+ assertEquals(nodeId, raftManager.client.nodeId.getAsInt)
raftManager.shutdown()
}
- @Test
- def testNodeIdPresentIfControllerRoleOnly(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "controller", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ @ParameterizedTest
+ @ValueSource(strings = Array("metadata", "log", "metadata,log"))
+ def testLogDirLockWhenControllerOnly(dirType: String): Unit = {
+ val logDir = if (dirType.contains("metadata")) {
+ Some(TestUtils.tempDir().toPath)
+ } else {
+ None
+ }
+
+ val metadataDir = if (dirType.contains("log")) {
+ Some(TestUtils.tempDir().toPath)
+ } else {
+ None
+ }
+
+ val nodeId = 1
+ val raftManager = createRaftManager(
+ new TopicPartition("__raft_id_test", 0),
+ createConfig(
+ Set(ControllerRole),
+ nodeId,
+ logDir,
+ metadataDir
+ )
+ )
+
+ val lockPath = metadataDir.getOrElse(logDir.get).resolve(LogManager.LockFileName)
+ assertTrue(fileLocked(lockPath))
+
raftManager.shutdown()
+
+ assertFalse(fileLocked(lockPath))
}
@Test
- def testNodeIdPresentIfColocated(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "controller,broker", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ def testLogDirLockWhenMetadataDir(): Unit = {
+ val logDir = Some(TestUtils.tempDir().toPath)
+ val metadataDir = Some(TestUtils.tempDir().toPath)
+
+ val nodeId = 1
+ val raftManager = createRaftManager(
+ new TopicPartition("__raft_id_test", 0),
+ createConfig(
+ Set(BrokerRole),
Review Comment:
I wanted to test having different `metadata.log.dir` and `log.dirs` with the broker. The matrix in `testLogDirLockWhenControllerOnly` already tests having different `metadata.log.dir` and `log.dirs` for the controller.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio merged pull request #13058: KAFKA-14557; Lock metadata log dir
Posted by GitBox <gi...@apache.org>.
jsancio merged PR #13058:
URL: https://github.com/apache/kafka/pull/13058
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a diff in pull request #13058: KAFKA-14557; Lock metadata log dir
Posted by GitBox <gi...@apache.org>.
rondagostino commented on code in PR #13058:
URL: https://github.com/apache/kafka/pull/13058#discussion_r1065242904
##########
core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala:
##########
@@ -127,18 +127,18 @@ class RaftManagerTest {
}
@ParameterizedTest
- @ValueSource(strings = Array("metadata", "log", "metadata,log"))
+ @ValueSource(strings = Array("metadata-only", "log-only", "both"))
def testLogDirLockWhenControllerOnly(dirType: String): Unit = {
- val logDir = if (dirType.contains("metadata")) {
- Some(TestUtils.tempDir().toPath)
- } else {
+ val logDir = if (dirType.equals("metadata-only")) {
None
+ } else {
+ Some(TestUtils.tempDir().toPath)
}
- val metadataDir = if (dirType.contains("log")) {
- Some(TestUtils.tempDir().toPath)
- } else {
+ val metadataDir = if (dirType.contains("log-only")) {
Review Comment:
`s/contains/equals/`
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a diff in pull request #13058: KAFKA-14557; Lock metadata log dir
Posted by GitBox <gi...@apache.org>.
rondagostino commented on code in PR #13058:
URL: https://github.com/apache/kafka/pull/13058#discussion_r1064790638
##########
core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala:
##########
@@ -81,25 +100,99 @@ class RaftManagerTest {
)
}
- @Test
- def testNodeIdPresentIfBrokerRoleOnly(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "broker", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ @ParameterizedTest
+ @ValueSource(strings = Array("broker", "controller", "broker,controller"))
+ def testNodeIdPresent(processRoles: String): Unit = {
+ var processRolesSet = Set.empty[ProcessRole]
+ if (processRoles.contains("broker")) {
+ processRolesSet = processRolesSet ++ Set(BrokerRole)
+ }
+ if (processRoles.contains("controller")) {
+ processRolesSet = processRolesSet ++ Set(ControllerRole)
+ }
+
+ val logDir = TestUtils.tempDir()
+ val nodeId = 1
+ val raftManager = createRaftManager(
+ new TopicPartition("__raft_id_test", 0),
+ createConfig(
+ processRolesSet,
+ nodeId,
+ Some(logDir.toPath),
+ None
+ )
+ )
+ assertEquals(nodeId, raftManager.client.nodeId.getAsInt)
raftManager.shutdown()
}
- @Test
- def testNodeIdPresentIfControllerRoleOnly(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "controller", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ @ParameterizedTest
+ @ValueSource(strings = Array("metadata", "log", "metadata,log"))
+ def testLogDirLockWhenControllerOnly(dirType: String): Unit = {
+ val logDir = if (dirType.contains("metadata")) {
+ Some(TestUtils.tempDir().toPath)
+ } else {
+ None
+ }
+
+ val metadataDir = if (dirType.contains("log")) {
+ Some(TestUtils.tempDir().toPath)
+ } else {
+ None
+ }
Review Comment:
I think this might be clearer.
```
@ValueSource(strings = Array("metadata-only", "log-only", "both"))
def testLogDirLockWhenControllerOnly(dirType: String): Unit = {
val logDir = if (!dirType.equals("metadata-only")) {
Some(TestUtils.tempDir().toPath)
} else {
None
}
val metadataDir = if (!dirType.equals("log-only")) {
Some(TestUtils.tempDir().toPath)
} else {
None
}
```
##########
core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala:
##########
@@ -81,25 +100,99 @@ class RaftManagerTest {
)
}
- @Test
- def testNodeIdPresentIfBrokerRoleOnly(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "broker", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ @ParameterizedTest
+ @ValueSource(strings = Array("broker", "controller", "broker,controller"))
+ def testNodeIdPresent(processRoles: String): Unit = {
+ var processRolesSet = Set.empty[ProcessRole]
+ if (processRoles.contains("broker")) {
+ processRolesSet = processRolesSet ++ Set(BrokerRole)
+ }
+ if (processRoles.contains("controller")) {
+ processRolesSet = processRolesSet ++ Set(ControllerRole)
+ }
+
+ val logDir = TestUtils.tempDir()
+ val nodeId = 1
+ val raftManager = createRaftManager(
+ new TopicPartition("__raft_id_test", 0),
+ createConfig(
+ processRolesSet,
+ nodeId,
+ Some(logDir.toPath),
+ None
+ )
+ )
+ assertEquals(nodeId, raftManager.client.nodeId.getAsInt)
raftManager.shutdown()
}
- @Test
- def testNodeIdPresentIfControllerRoleOnly(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "controller", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ @ParameterizedTest
+ @ValueSource(strings = Array("metadata", "log", "metadata,log"))
+ def testLogDirLockWhenControllerOnly(dirType: String): Unit = {
+ val logDir = if (dirType.contains("metadata")) {
+ Some(TestUtils.tempDir().toPath)
+ } else {
+ None
+ }
+
+ val metadataDir = if (dirType.contains("log")) {
+ Some(TestUtils.tempDir().toPath)
+ } else {
+ None
+ }
+
+ val nodeId = 1
+ val raftManager = createRaftManager(
+ new TopicPartition("__raft_id_test", 0),
+ createConfig(
+ Set(ControllerRole),
+ nodeId,
+ logDir,
+ metadataDir
+ )
+ )
+
+ val lockPath = metadataDir.getOrElse(logDir.get).resolve(LogManager.LockFileName)
+ assertTrue(fileLocked(lockPath))
+
raftManager.shutdown()
+
+ assertFalse(fileLocked(lockPath))
}
@Test
- def testNodeIdPresentIfColocated(): Unit = {
- val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "controller,broker", "1")
- assertEquals(1, raftManager.client.nodeId.getAsInt)
+ def testLogDirLockWhenMetadataDir(): Unit = {
Review Comment:
A better name might be `testLogDirLockWhenBrokerOnlyWithSeparateMetadataDir`
##########
core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala:
##########
@@ -16,52 +16,71 @@
*/
package kafka.raft
-import java.util.concurrent.CompletableFuture
+import java.nio.channels.FileChannel
+import java.nio.channels.OverlappingFileLockException
+import java.nio.file.Path
+import java.nio.file.StandardOpenOption
import java.util.Properties
+import java.util.concurrent.CompletableFuture
+import kafka.log.LogManager
import kafka.raft.KafkaRaftManager.RaftIoThread
-import kafka.server.{KafkaConfig, MetaProperties}
+import kafka.server.KafkaConfig
+import kafka.server.KafkaRaftServer.BrokerRole
+import kafka.server.KafkaRaftServer.ControllerRole
+import kafka.server.KafkaRaftServer.ProcessRole
+import kafka.server.MetaProperties
+import kafka.utils.TestUtils
import kafka.tools.TestRaftServer.ByteArraySerde
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.Uuid
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.utils.Time
import org.apache.kafka.raft.KafkaRaftClient
import org.apache.kafka.raft.RaftConfig
-import org.apache.kafka.test.TestUtils
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
import org.mockito.Mockito._
-import java.io.File
-
class RaftManagerTest {
-
- private def instantiateRaftManagerWithConfigs(topicPartition: TopicPartition, processRoles: String, nodeId: String) = {
- def configWithProcessRolesAndNodeId(processRoles: String, nodeId: String, logDir: File): KafkaConfig = {
- val props = new Properties
- props.setProperty(KafkaConfig.MetadataLogDirProp, logDir.getPath)
- props.setProperty(KafkaConfig.ProcessRolesProp, processRoles)
- props.setProperty(KafkaConfig.NodeIdProp, nodeId)
- props.setProperty(KafkaConfig.ControllerListenerNamesProp, "SSL")
- if (processRoles.contains("broker")) {
- props.setProperty(KafkaConfig.InterBrokerListenerNameProp, "PLAINTEXT")
- if (processRoles.contains("controller")) { // co-located
- props.setProperty(KafkaConfig.ListenersProp, "PLAINTEXT://localhost:9092,SSL://localhost:9093")
- props.setProperty(KafkaConfig.QuorumVotersProp, s"${nodeId}@localhost:9093")
- } else { // broker-only
- val voterId = (nodeId.toInt + 1)
- props.setProperty(KafkaConfig.QuorumVotersProp, s"${voterId}@localhost:9093")
- }
- } else if (processRoles.contains("controller")) { // controller-only
- props.setProperty(KafkaConfig.ListenersProp, "SSL://localhost:9093")
+ private def createConfig(
+ processRoles: Set[ProcessRole],
+ nodeId: Int,
+ logDir: Option[Path],
+ metadataDir: Option[Path]
+ ): KafkaConfig = {
+ val props = new Properties
+ logDir.foreach { value =>
+ props.setProperty(KafkaConfig.LogDirProp, value.toString)
+ }
+ metadataDir.foreach { value =>
+ props.setProperty(KafkaConfig.MetadataLogDirProp, value.toString)
+ }
+ props.setProperty(KafkaConfig.ProcessRolesProp, processRoles.mkString(","))
+ props.setProperty(KafkaConfig.NodeIdProp, nodeId.toString)
+ props.setProperty(KafkaConfig.ControllerListenerNamesProp, "SSL")
+ if (processRoles.contains(BrokerRole)) {
+ props.setProperty(KafkaConfig.InterBrokerListenerNameProp, "PLAINTEXT")
+ if (processRoles.contains(ControllerRole)) { // co-located
+ props.setProperty(KafkaConfig.ListenersProp, "PLAINTEXT://localhost:9092,SSL://localhost:9093")
props.setProperty(KafkaConfig.QuorumVotersProp, s"${nodeId}@localhost:9093")
+ } else { // broker-only
+ val voterId = (nodeId.toInt + 1)
Review Comment:
`s/(nodeId.toInt + 1)/nodeId + 1/`
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org