You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by GitBox <gi...@apache.org> on 2020/11/04 22:08:44 UTC
[GitHub] [samza] MabelYC opened a new pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
MabelYC opened a new pull request #1437:
URL: https://github.com/apache/samza/pull/1437
Symptom: When deploying a job, users will see error messages from the kafka clients finalizer complaining about "kafka consumer/producer allocated and not closed"
These error messages appear multiple times in users' jobs and don't actually affect the job's performance or correctness. This leads to users falsely believing that any job failures is because of this message.
Cause:
All Kafka clients have a finalize method in them, when GC determines that there's no more references to the clients, it will try to close the clients and find that the clients wasn't closed properly yet. And was caused by two different reasons:
1. In ChangelogStreamManager.class and RegexTopicGenerator.class, Samza created admin Clients for every system, no matter we need them or not. But we only stop the admin clients that we used.
2. In KafkaCheckpointManager.class and CoordinatorStreamStore.class, we want to keep the kafka clients open until the containers are shut down. But during runtime, the ERROR log would show when there's no more references to the clients.
Changes:
1. In ChangelogStreamManager.class and RegexTopicGenerator.class, we will only create adminClients that we needed instead of creating adminClients for all systems and then pick up the one we want.
2. In KafkaCheckpointManager.class and CoordinatorStreamStore.class, we override a finalize method to close the clients if there's no more references to the clients.
Tests: local deployed some test-jobs including samza and beam-samza jobs, the error logs don't show again.
API Changes: N/A
Upgrade Instructions: N/A
Usage Instructions:N/A
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] MabelYC commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
MabelYC commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r528043170
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -441,15 +441,24 @@ public void afterStop() {
@Override
public void afterFailure(Throwable t) {
- processors.removeIf(pair -> pair.getLeft().equals(processor));
-
+ // we need to close associated coordinator metadata store, although the processor failed
+ processors.forEach(sp -> {
Review comment:
I think metadata store are closed before calling afterStop().
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] MabelYC commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
MabelYC commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r526535064
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
##########
@@ -111,6 +111,12 @@ public void init() {
systemConsumer.start();
systemProducer.register(SOURCE);
systemProducer.start();
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ LOG.info("CoordinatorStreamStore Shut Down Hook thread is closing kafka clients");
+ this.systemProducer.stop();
Review comment:
changed
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] sborya commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
sborya commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r520889344
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
##########
@@ -188,6 +188,14 @@ public void flush() {
}
}
+ @Override
+ protected void finalize() throws Throwable {
Review comment:
I don't think using finalize is a good idea.
This will cause unpredictable performance impact, since each garbage collection may take an arbitrary time now.
The services should be stopped as a part of stopped flow, not on garbage collection.
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
##########
@@ -188,6 +188,14 @@ public void flush() {
}
}
+ @Override
Review comment:
preferred way of calling super.finalize should be:
finalize() {
try {
.. stop...
} finally {
super.finalize();
}
}
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] kw2542 commented on pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
kw2542 commented on pull request #1437:
URL: https://github.com/apache/samza/pull/1437#issuecomment-729450893
out of curiosity, is it possible that the created kafka clients are being garbage collected before shutdown hooks executed?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] kw2542 commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r525832870
##########
File path: samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
##########
@@ -112,6 +115,12 @@ class KafkaCheckpointManager(checkpointSpec: KafkaStreamSpec,
info(s"Starting the checkpoint SystemConsumer from oldest offset $oldestOffset")
systemConsumer.register(checkpointSsp, oldestOffset)
systemConsumer.start()
+ Runtime.getRuntime.addShutdownHook(new Thread("KafkaCheckPointManagerShutdownHook") {
+ override def run(): Unit = {
+ producerRef.get().stop()
Review comment:
this.stop()?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] prateekm commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
prateekm commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r527227011
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
##########
@@ -111,6 +111,10 @@ public void init() {
systemConsumer.start();
systemProducer.register(SOURCE);
systemProducer.start();
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
Review comment:
Relying on shutdown hooks for lifecycle management of components is not a good idea.
The close should be called by whoever created the SystemProducer etc instead. If it's created in this class, this class should have lifecycle methods of it's own, which should be called by the owner of this class.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] kw2542 commented on pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
kw2542 commented on pull request #1437:
URL: https://github.com/apache/samza/pull/1437#issuecomment-722034813
LGTM
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] kw2542 commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r527258814
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
##########
@@ -111,6 +111,10 @@ public void init() {
systemConsumer.start();
systemProducer.register(SOURCE);
systemProducer.start();
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
Review comment:
+1, shutdown hooks are being executed when JVM exists, it does not seem to fit this scenario. It could be less appropriate compared to finalize() as the warning messages we get from Kafka is emitted in finalize() as well.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] prateekm commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
prateekm commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r528027101
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -441,15 +441,24 @@ public void afterStop() {
@Override
public void afterFailure(Throwable t) {
- processors.removeIf(pair -> pair.getLeft().equals(processor));
-
+ // we need to close associated coordinator metadata store, although the processor failed
+ processors.forEach(sp -> {
+ if (sp.getLeft().equals(processor)) {
+ if (sp.getRight() != null) {
+ sp.getRight().close();
+ }
+ processors.remove(sp);
Review comment:
Is it safe to modify (remove) the map here while iterating over it (foreach)?
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -441,15 +441,24 @@ public void afterStop() {
@Override
public void afterFailure(Throwable t) {
- processors.removeIf(pair -> pair.getLeft().equals(processor));
-
+ // we need to close associated coordinator metadata store, although the processor failed
+ processors.forEach(sp -> {
Review comment:
Won't this cause the current processor to not be stopped?
Do we also need to stop the metadata store in afterStop()?
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/MetadataResourceUtil.java
##########
@@ -55,6 +55,7 @@ public MetadataResourceUtil(JobModel jobModel, MetricsRegistry metricsRegistry,
public void createResources() {
if (checkpointManager != null) {
checkpointManager.createResources();
+ checkpointManager.stop();
Review comment:
Do we know for sure that this method is only called once? If so, do we need to create the checkpoint manager during construction, or can we create it and close it within createResources?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] MabelYC commented on pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
MabelYC commented on pull request #1437:
URL: https://github.com/apache/samza/pull/1437#issuecomment-730067476
> garbage collected before shutdown hooks executed
That's a really interesting question. Let me do some search about it.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] prateekm commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
prateekm commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r527303445
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
##########
@@ -111,6 +111,10 @@ public void init() {
systemConsumer.start();
systemProducer.register(SOURCE);
systemProducer.start();
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
Review comment:
@kw2542 Just FYI, @MabelYC and I were able to identify the places where we're leaking references in JobCoordinator and LocalApplicationRunner.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] MabelYC commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
MabelYC commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r530061841
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -441,15 +441,24 @@ public void afterStop() {
@Override
public void afterFailure(Throwable t) {
- processors.removeIf(pair -> pair.getLeft().equals(processor));
-
+ // we need to close associated coordinator metadata store, although the processor failed
+ processors.forEach(sp -> {
Review comment:
I double checked that the close loop contains some LI specific hookup with offspring. I updated the commit to stop the metadata store in afterStop() to make sure it got closed for sure.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] kw2542 commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r527326529
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
##########
@@ -111,6 +111,10 @@ public void init() {
systemConsumer.start();
systemProducer.register(SOURCE);
systemProducer.start();
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
Review comment:
This is awesome!
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] kw2542 commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
kw2542 commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r525832466
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
##########
@@ -111,6 +111,12 @@ public void init() {
systemConsumer.start();
systemProducer.register(SOURCE);
systemProducer.start();
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ LOG.info("CoordinatorStreamStore Shut Down Hook thread is closing kafka clients");
+ this.systemProducer.stop();
Review comment:
can we do this.close() ?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] prateekm commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
prateekm commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r529787680
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/MetadataResourceUtil.java
##########
@@ -55,6 +55,7 @@ public MetadataResourceUtil(JobModel jobModel, MetricsRegistry metricsRegistry,
public void createResources() {
if (checkpointManager != null) {
checkpointManager.createResources();
+ checkpointManager.stop();
Review comment:
@MabelYC Sorry, don't understand the first part of your reply. What is that in response to?
Creating the checkpoint manager in the constructor and stopping it after the first use is :
1. Asymmetrical. Ideally this class should have lifecycle methods (start/stop) and the checkpoint manager would be stopped there. Also, looks like we don't even start the checkpoint manager?
2. Makes an unsafe assumption that the method is only called once. If it was ever called more than once, the second call would fail since the checkpoint manager is already stopped.
IMHO it will be cleaner to create a new checkpoint manager in createResources and close it after use. That way there is no additional overhead as long as it's called once, and its safe if its called more than once.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] MabelYC commented on pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
MabelYC commented on pull request #1437:
URL: https://github.com/apache/samza/pull/1437#issuecomment-729315648
> I don't like using finalize() in objects. We ran into huge GC issues because of it in the past.
> You can search google for this topic (e.g. https://howtodoinjava.com/java-examples/why-not-to-use-finalize-method-in-java/).
> Please consider alternative solutions.
Changed to use addShutdownHook to close clients. Please take a look.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] prateekm commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
prateekm commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r527265232
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
##########
@@ -111,6 +111,10 @@ public void init() {
systemConsumer.start();
systemProducer.register(SOURCE);
systemProducer.start();
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
Review comment:
@kw2542 We shouldn't use finalize either. The Kafka warning shows that we're leaking client references somewhere without closing them. We should try to find and fix the underlying object lifecycle issue first instead of fixing the symptom.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] MabelYC commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
MabelYC commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r528027976
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
##########
@@ -111,6 +111,10 @@ public void init() {
systemConsumer.start();
systemProducer.register(SOURCE);
systemProducer.start();
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
Review comment:
@kw2542 @prateekm @sborya Changed to close the clients according to their own lifecycles. Please take a 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] sborya merged pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
sborya merged pull request #1437:
URL: https://github.com/apache/samza/pull/1437
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] prateekm commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
prateekm commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r529790732
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -441,15 +441,24 @@ public void afterStop() {
@Override
public void afterFailure(Throwable t) {
- processors.removeIf(pair -> pair.getLeft().equals(processor));
-
+ // we need to close associated coordinator metadata store, although the processor failed
+ processors.forEach(sp -> {
Review comment:
@MabelYC It's not obvious to me how we can guarantee that all metadata stores are always closed. Can you document why you think metadata stores are always closed before calling afterStop?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] MabelYC commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
MabelYC commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r530062128
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/MetadataResourceUtil.java
##########
@@ -55,6 +55,7 @@ public MetadataResourceUtil(JobModel jobModel, MetricsRegistry metricsRegistry,
public void createResources() {
if (checkpointManager != null) {
checkpointManager.createResources();
+ checkpointManager.stop();
Review comment:
Got it. Updated the pr. Created an adminClient for this function.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [samza] MabelYC commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
MabelYC commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r526535509
##########
File path: samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
##########
@@ -112,6 +115,12 @@ class KafkaCheckpointManager(checkpointSpec: KafkaStreamSpec,
info(s"Starting the checkpoint SystemConsumer from oldest offset $oldestOffset")
systemConsumer.register(checkpointSsp, oldestOffset)
systemConsumer.start()
+ Runtime.getRuntime.addShutdownHook(new Thread("KafkaCheckPointManagerShutdownHook") {
+ override def run(): Unit = {
+ producerRef.get().stop()
Review comment:
adminClient and producer/consumer are not started at the same time. so I think it is better to add different shutdownhook 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] MabelYC commented on a change in pull request #1437: SAMZA-2598:Kafka clients are not created and closed properly
Posted by GitBox <gi...@apache.org>.
MabelYC commented on a change in pull request #1437:
URL: https://github.com/apache/samza/pull/1437#discussion_r528043358
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/MetadataResourceUtil.java
##########
@@ -55,6 +55,7 @@ public MetadataResourceUtil(JobModel jobModel, MetricsRegistry metricsRegistry,
public void createResources() {
if (checkpointManager != null) {
checkpointManager.createResources();
+ checkpointManager.stop();
Review comment:
I also thought we could create an admin client only to create resources. But i think it is not that good that then we would create two kafka adminClients for the same system in the same manager, and also create another thread. I think it would be better if we only create the clients we really need to create.
I double checked the method is only called once. And advantages not to create it during construction?
##########
File path: samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
##########
@@ -441,15 +441,24 @@ public void afterStop() {
@Override
public void afterFailure(Throwable t) {
- processors.removeIf(pair -> pair.getLeft().equals(processor));
-
+ // we need to close associated coordinator metadata store, although the processor failed
+ processors.forEach(sp -> {
+ if (sp.getLeft().equals(processor)) {
+ if (sp.getRight() != null) {
+ sp.getRight().close();
+ }
+ processors.remove(sp);
Review comment:
Thanks for pointing this. Changed to remove it after finishing iterating.
----------------------------------------------------------------
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:
users@infra.apache.org