You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "showuon (via GitHub)" <gi...@apache.org> on 2023/12/27 08:36:39 UTC
[PR] MINOR: close leaking threads in replica manager tests [kafka]
showuon opened a new pull request, #15077:
URL: https://github.com/apache/kafka/pull/15077
Following @dajac 's finding in https://github.com/apache/kafka/pull/15063, I found we also create new RemoteLogManager in ReplicaManagerTest, but didn't close them.
While investigating ReplicaManagerTest, I also found there are other threads leaking:
1. remote fetch reaper thread. It's because we create a reaper thread in test, which is not expected. We should create a mocked one like other purgatory instance.
2. Throttle threads. We created a `quotaManager` to feed into the replicaManager, but didn't close it. Actually, we have created a global `quotaManager` instance and will close it on `AfterEach`. We should re-use it.
### 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
Re: [PR] KAFKA-16074: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15077:
URL: https://github.com/apache/kafka/pull/15077#issuecomment-1876239029
@satishd , please take another look when available. Thanks.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1437350376
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2704,28 +2710,30 @@ class ReplicaManagerTest {
time = time,
scheduler = time.scheduler,
logManager = logManager,
- quotaManagers = QuotaFactory.instantiate(config, metrics, time, ""),
+ quotaManagers = quotaManager,
metadataCache = MetadataCache.zkMetadataCache(config.brokerId, config.interBrokerProtocolVersion),
logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
alterPartitionManager = alterPartitionManager,
threadNamePrefix = Option(this.getClass.getName))
- logManager.startup(Set.empty[String])
-
- // Create a hosted topic, a hosted topic that will become stray, and a stray topic
- val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
- createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
- createStrayLogs(10, logManager)
+ try {
+ logManager.startup(Set.empty[String])
- val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
+ // Create a hosted topic, a hosted topic that will become stray, and a stray topic
+ val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
+ createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
+ createStrayLogs(10, logManager)
- replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
+ val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
- assertEquals(validLogs, logManager.allLogs.toSet)
- assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
- replicaManager.shutdown()
- logManager.shutdown()
+ assertEquals(validLogs, logManager.allLogs.toSet)
+ assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ } finally {
+ replicaManager.shutdown()
+ logManager.shutdown()
Review Comment:
I've updated the PR. Thanks.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1437026129
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2704,28 +2710,30 @@ class ReplicaManagerTest {
time = time,
scheduler = time.scheduler,
logManager = logManager,
- quotaManagers = QuotaFactory.instantiate(config, metrics, time, ""),
+ quotaManagers = quotaManager,
metadataCache = MetadataCache.zkMetadataCache(config.brokerId, config.interBrokerProtocolVersion),
logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
alterPartitionManager = alterPartitionManager,
threadNamePrefix = Option(this.getClass.getName))
- logManager.startup(Set.empty[String])
-
- // Create a hosted topic, a hosted topic that will become stray, and a stray topic
- val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
- createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
- createStrayLogs(10, logManager)
+ try {
+ logManager.startup(Set.empty[String])
- val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
+ // Create a hosted topic, a hosted topic that will become stray, and a stray topic
+ val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
+ createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
+ createStrayLogs(10, logManager)
- replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
+ val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
- assertEquals(validLogs, logManager.allLogs.toSet)
- assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
- replicaManager.shutdown()
- logManager.shutdown()
+ assertEquals(validLogs, logManager.allLogs.toSet)
+ assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ } finally {
+ replicaManager.shutdown()
+ logManager.shutdown()
Review Comment:
So, I think I should only swallow in more than 1 instances to get close in finally block. Is that what you expect @satishd ?
--
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
Re: [PR] KAFKA-16074: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1441491733
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -4115,7 +4131,8 @@ class ReplicaManagerTest {
safeYammerMetricValue("type=DelayedRemoteFetchMetrics,name=ExpiresPerSec").asInstanceOf[Long])
latch.countDown()
} finally {
- replicaManager.shutdown(checkpointHW = false)
+ CoreUtils.swallow(replicaManager.shutdown(checkpointHW = false), this)
Review Comment:
I agree. Will update it soon.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1437344252
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -167,7 +167,7 @@ class ReplicaManagerTest {
.foreach(checkpointFile => assertTrue(Files.exists(checkpointFile),
s"checkpoint file does not exist at $checkpointFile"))
} finally {
- rm.shutdown(checkpointHW = false)
+ CoreUtils.swallow(rm.shutdown(checkpointHW = false), this)
Review Comment:
Why is the exception swallowed here and all other places with the latest change? I did not mean to suggest this in my review. I clarified that in another [comment](https://github.com/apache/kafka/pull/15077#discussion_r1437341409).
--
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
Re: [PR] KAFKA-16074: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon merged PR #15077:
URL: https://github.com/apache/kafka/pull/15077
--
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
Re: [PR] KAFKA-16074: close leaking threads in replica manager tests [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1441456492
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -4115,7 +4131,8 @@ class ReplicaManagerTest {
safeYammerMetricValue("type=DelayedRemoteFetchMetrics,name=ExpiresPerSec").asInstanceOf[Long])
latch.countDown()
} finally {
- replicaManager.shutdown(checkpointHW = false)
+ CoreUtils.swallow(replicaManager.shutdown(checkpointHW = false), this)
Review Comment:
With the suggested change of using [Utils.tryAll()](https://github.com/apache/kafka/blob/810f63b5b645666c004a6a1222eeed54b202c10e/clients/src/main/java/org/apache/kafka/common/utils/Utils.java#L1197) instead of swallowing exceptions, it becomes
```
Utils.tryAll(util.Arrays.asList[Callable[Void]](
() => {
replicaManager.shutdown(checkpointHW = false)
null
},
() => {
remoteLogManager.close()
null
},
))
```
--
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
Re: [PR] KAFKA-16074: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15077:
URL: https://github.com/apache/kafka/pull/15077#issuecomment-1876745554
@satishd , PR updated. Thanks.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1436993064
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -4020,6 +4031,7 @@ class ReplicaManagerTest {
doneLatch.countDown()
} finally {
replicaManager.shutdown(checkpointHW = false)
+ remoteLogManager.close()
Review Comment:
Can you close/shutdown safely by swallowing any intermediate exceptions occur as mentioned below?
```
CoreUtils.swallow(replicaManager.shutdown(checkpointHW = false), this)
CoreUtils.swallow(remoteLogManager.close(), this)
```
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2704,28 +2710,30 @@ class ReplicaManagerTest {
time = time,
scheduler = time.scheduler,
logManager = logManager,
- quotaManagers = QuotaFactory.instantiate(config, metrics, time, ""),
+ quotaManagers = quotaManager,
metadataCache = MetadataCache.zkMetadataCache(config.brokerId, config.interBrokerProtocolVersion),
logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
alterPartitionManager = alterPartitionManager,
threadNamePrefix = Option(this.getClass.getName))
- logManager.startup(Set.empty[String])
-
- // Create a hosted topic, a hosted topic that will become stray, and a stray topic
- val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
- createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
- createStrayLogs(10, logManager)
+ try {
+ logManager.startup(Set.empty[String])
- val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
+ // Create a hosted topic, a hosted topic that will become stray, and a stray topic
+ val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
+ createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
+ createStrayLogs(10, logManager)
- replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
+ val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
- assertEquals(validLogs, logManager.allLogs.toSet)
- assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
- replicaManager.shutdown()
- logManager.shutdown()
+ assertEquals(validLogs, logManager.allLogs.toSet)
+ assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ } finally {
+ replicaManager.shutdown()
+ logManager.shutdown()
Review Comment:
Can you close both of them safely like below ignoring any intermediate failures?
```
CoreUtils.swallow(replicaManager.shutdown(checkpointHW = false), this)
CoreUtils.swallow(remoteLogManager.close(), this)
```
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -4116,6 +4128,7 @@ class ReplicaManagerTest {
latch.countDown()
} finally {
replicaManager.shutdown(checkpointHW = false)
+ remoteLogManager.close()
Review Comment:
Please close safely as mentioned in the earlier comment.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1437034332
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2704,28 +2710,30 @@ class ReplicaManagerTest {
time = time,
scheduler = time.scheduler,
logManager = logManager,
- quotaManagers = QuotaFactory.instantiate(config, metrics, time, ""),
+ quotaManagers = quotaManager,
metadataCache = MetadataCache.zkMetadataCache(config.brokerId, config.interBrokerProtocolVersion),
logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
alterPartitionManager = alterPartitionManager,
threadNamePrefix = Option(this.getClass.getName))
- logManager.startup(Set.empty[String])
-
- // Create a hosted topic, a hosted topic that will become stray, and a stray topic
- val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
- createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
- createStrayLogs(10, logManager)
+ try {
+ logManager.startup(Set.empty[String])
- val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
+ // Create a hosted topic, a hosted topic that will become stray, and a stray topic
+ val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
+ createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
+ createStrayLogs(10, logManager)
- replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
+ val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
- assertEquals(validLogs, logManager.allLogs.toSet)
- assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
- replicaManager.shutdown()
- logManager.shutdown()
+ assertEquals(validLogs, logManager.allLogs.toSet)
+ assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ } finally {
+ replicaManager.shutdown()
+ logManager.shutdown()
Review Comment:
That is fair. In that case, we should perhaps use `public static void closeQuietly(AutoCloseable closeable, String name, AtomicReference<Throwable> firstException)` and end the finally with something similar to
```
if (firstException != null) {
throw new KafkaException("Failed to close", exception);
}
```
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15077:
URL: https://github.com/apache/kafka/pull/15077#issuecomment-1870263212
@satishd , I've closed all the instances in finally block quietly in this commit: https://github.com/apache/kafka/pull/15077/commits/dd913a8668cf773a51403e482cc704b44cb0e8a1 . Thanks.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1437022630
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2704,28 +2710,30 @@ class ReplicaManagerTest {
time = time,
scheduler = time.scheduler,
logManager = logManager,
- quotaManagers = QuotaFactory.instantiate(config, metrics, time, ""),
+ quotaManagers = quotaManager,
metadataCache = MetadataCache.zkMetadataCache(config.brokerId, config.interBrokerProtocolVersion),
logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
alterPartitionManager = alterPartitionManager,
threadNamePrefix = Option(this.getClass.getName))
- logManager.startup(Set.empty[String])
-
- // Create a hosted topic, a hosted topic that will become stray, and a stray topic
- val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
- createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
- createStrayLogs(10, logManager)
+ try {
+ logManager.startup(Set.empty[String])
- val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
+ // Create a hosted topic, a hosted topic that will become stray, and a stray topic
+ val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
+ createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
+ createStrayLogs(10, logManager)
- replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
+ val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
- assertEquals(validLogs, logManager.allLogs.toSet)
- assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
- replicaManager.shutdown()
- logManager.shutdown()
+ assertEquals(validLogs, logManager.allLogs.toSet)
+ assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ } finally {
+ replicaManager.shutdown()
+ logManager.shutdown()
Review Comment:
Do we want to close them quietly? This will suppress memory leaks since we won't get to know if the close is not happening correctly. It's better to fail when close has an error so that we get to know that something is wrong with tests,
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1436885274
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2693,6 +2693,9 @@ class ReplicaManagerTest {
} else {
assertTrue(stray0.isInstanceOf[HostedPartition.Online])
}
+
+ replicaManager.shutdown()
+ logManager.shutdown()
Review Comment:
we should probably do it in "finally" otherwise we we may not close them if assertion fails or an exception is thrown
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1437157116
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2639,59 +2639,65 @@ class ReplicaManagerTest {
time = time,
scheduler = time.scheduler,
logManager = logManager,
- quotaManagers = QuotaFactory.instantiate(config, metrics, time, ""),
Review Comment:
Was there a reason we created a separated one here? I noticed that we change the config above if zkMigration is enabled, but not sure if that makes a difference.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1437350213
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2639,59 +2639,65 @@ class ReplicaManagerTest {
time = time,
scheduler = time.scheduler,
logManager = logManager,
- quotaManagers = QuotaFactory.instantiate(config, metrics, time, ""),
Review Comment:
I agree we should not change any test semantics here. I updated the PR to close the new created `quotaManager` explicitly.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1437341409
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2704,28 +2710,30 @@ class ReplicaManagerTest {
time = time,
scheduler = time.scheduler,
logManager = logManager,
- quotaManagers = QuotaFactory.instantiate(config, metrics, time, ""),
+ quotaManagers = quotaManager,
metadataCache = MetadataCache.zkMetadataCache(config.brokerId, config.interBrokerProtocolVersion),
logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
alterPartitionManager = alterPartitionManager,
threadNamePrefix = Option(this.getClass.getName))
- logManager.startup(Set.empty[String])
-
- // Create a hosted topic, a hosted topic that will become stray, and a stray topic
- val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
- createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
- createStrayLogs(10, logManager)
+ try {
+ logManager.startup(Set.empty[String])
- val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
+ // Create a hosted topic, a hosted topic that will become stray, and a stray topic
+ val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
+ createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
+ createStrayLogs(10, logManager)
- replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
+ val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
- assertEquals(validLogs, logManager.allLogs.toSet)
- assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
- replicaManager.shutdown()
- logManager.shutdown()
+ assertEquals(validLogs, logManager.allLogs.toSet)
+ assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ } finally {
+ replicaManager.shutdown()
+ logManager.shutdown()
Review Comment:
Right @showuon , what I meant here was try closing the second one even if the earlier invocation has thrown an error to avoid leaks caused by not closing the second instance. We should not change any of the other places to swallow the exceptions where there is only one single invocation of shutting down of replicamanager or logmanager.
One way to do that is what @divijvaidya suggested.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1437024347
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2704,28 +2710,30 @@ class ReplicaManagerTest {
time = time,
scheduler = time.scheduler,
logManager = logManager,
- quotaManagers = QuotaFactory.instantiate(config, metrics, time, ""),
+ quotaManagers = quotaManager,
metadataCache = MetadataCache.zkMetadataCache(config.brokerId, config.interBrokerProtocolVersion),
logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
alterPartitionManager = alterPartitionManager,
threadNamePrefix = Option(this.getClass.getName))
- logManager.startup(Set.empty[String])
-
- // Create a hosted topic, a hosted topic that will become stray, and a stray topic
- val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
- createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
- createStrayLogs(10, logManager)
+ try {
+ logManager.startup(Set.empty[String])
- val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
+ // Create a hosted topic, a hosted topic that will become stray, and a stray topic
+ val validLogs = createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
+ createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
+ createStrayLogs(10, logManager)
- replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
+ val allReplicasFromLISR = Set(new TopicPartition("hosted-topic", 0), new TopicPartition("hosted-topic", 1))
- assertEquals(validLogs, logManager.allLogs.toSet)
- assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ replicaManager.updateStrayLogs(replicaManager.findStrayPartitionsFromLeaderAndIsr(allReplicasFromLISR))
- replicaManager.shutdown()
- logManager.shutdown()
+ assertEquals(validLogs, logManager.allLogs.toSet)
+ assertEquals(validLogs.size, replicaManager.partitionCount.value)
+ } finally {
+ replicaManager.shutdown()
+ logManager.shutdown()
Review Comment:
I think what Satish wants is we close both of them without throwing any exception to not closing the 2nd instance.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15077:
URL: https://github.com/apache/kafka/pull/15077#issuecomment-1870075510
@dajac @jolshan @satishd , call for review. Thanks.
--
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
Re: [PR] KAFKA-16074: close leaking threads in replica manager tests [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1441563341
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -4115,7 +4131,8 @@ class ReplicaManagerTest {
safeYammerMetricValue("type=DelayedRemoteFetchMetrics,name=ExpiresPerSec").asInstanceOf[Long])
latch.countDown()
} finally {
- replicaManager.shutdown(checkpointHW = false)
+ CoreUtils.swallow(replicaManager.shutdown(checkpointHW = false), this)
Review Comment:
Ah nice utility. TIL.
--
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
Re: [PR] MINOR: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15077:
URL: https://github.com/apache/kafka/pull/15077#discussion_r1436972102
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2693,6 +2693,9 @@ class ReplicaManagerTest {
} else {
assertTrue(stray0.isInstanceOf[HostedPartition.Online])
}
+
+ replicaManager.shutdown()
+ logManager.shutdown()
Review Comment:
You're right! Updated.
--
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
Re: [PR] KAFKA-16074: close leaking threads in replica manager tests [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15077:
URL: https://github.com/apache/kafka/pull/15077#issuecomment-1884704960
Failed tests are unrelated.
--
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