Wip/forklift kafka incremental controller #132
Conversation
| if (controller == null || !controller.isRunning()) { | ||
| controller = createController(); | ||
| KafkaController controller = controllers.get(name); | ||
| if(controller != null && controller.isRunning()){ |
| }); | ||
| // Start the consumer. | ||
| c1.listen(); | ||
| Thread.sleep(3000); |
There was a problem hiding this comment.
Do you need this sleep? I'm fairly certain that Consumer.listen is blocking.
There was a problem hiding this comment.
Yes, the sleep is to ensure that the consumer shuts down before the next consumer starts up.
There was a problem hiding this comment.
I see... the consumer is done processing messages, but you want to make sure that the close() is done being called on the previous KafkaTopicConsumer before getting a new KafkaTopicConsumer to read from, because otherwise you could make the new consumer using the old controller (and old MessageStream)?
There was a problem hiding this comment.
Yes. Now that we can't add and remove from the same controller this test is a little less relevant although I think it's worth leaving it in.
There was a problem hiding this comment.
Thinking some more, I guess I'd blame this being weird on Consumer.shutdown being a signal rather than an synchronous action that completes when the consumer is truly "shutdown" or a asynchronous task with a CompletableFuture. A better name for what it's doing is signalShutdown. 🤷♂️
|
Assuming this works, this should remove the need for your other bugfix PR? |
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.atomic.AtomicInteger; |
There was a problem hiding this comment.
Nit: the added imports don't appear to be used
| running = false; | ||
| kafkaConsumer.wakeup(); | ||
| // executor.shutdownNow(); | ||
| executor.shutdown(); |
There was a problem hiding this comment.
Do we even need shutdown here? The kafkaConsumer.wakeup() call should be sufficient to make the message loop thread terminate.
The difference between the two shutdown methods is that shutdown prevents incoming tasks from being queued on the executor (which shouldn't matter here since you only call start once on a KafkaController), while shutdown sends a signal Thread.interrupt to it's threads to let them know that they should quit soon.
There was a problem hiding this comment.
I think you meant shutdownNow causes an interrupt? I originally thought that I wouldn't need either of the shutdown calls as well when I did this refactor, but discovered that the awaitTermination requires a shutdown call of some sort to function. I'll remove the commented shutdownNow
There was a problem hiding this comment.
Yeah, whoops.
Hmm, maybe the designers wanted to be able to be sure that when awaitTermination is called that no new threads are going to be executed (because that makes the check more difficult than just a bunch of Thread.join calls). Seems good then.
| @@ -184,6 +141,10 @@ private void controlLoop() { | |||
| throw t; | |||
| } finally { | |||
| running = false; | |||
| synchronized (this) { | |||
| //wake up any threads waiting on a control loop operation | |||
| this.notifyAll(); | |||
There was a problem hiding this comment.
I don't see any waits to be woken up by this notify?
There was a problem hiding this comment.
Good catch, with the removal of the add/remove topic methods this is no longer needed.
| sentMessageIds.add(producer.send("message1")); | ||
| final Consumer c1 = new Consumer(StringConsumer.class, forklift); | ||
| // Shutdown the consumer after all the messages have been processed. | ||
| c1.setOutOfMessages((listener) -> { |
There was a problem hiding this comment.
Hopefully sending just one message doesn't need timeouts for CI... 🤞
|
Wait, what about supporting the new The difference is that you can only reuse the given |
Overview
Bridger discovered that calls to subscribe reset the fetch positions of topics. Additionally, there are some concerns that the topic partitions assigned for topics which do not change from subscription to subscription may be modified if the subscription occurs during some form of rebalance event. Although it should be possible to keep the existing approach, it was much simpler to simply move to a controller per topic setup. We may wish to re-visit this in the future should performance become an issue.