cadonna commented on code in PR #13318:
URL: https://github.com/apache/kafka/pull/13318#discussion_r1150536829
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -140,16 +134,15 @@ public void run() {
log.info("State updater thread started");
try {
while (isRunning.get()) {
- try {
- runOnce();
- } catch (final InterruptedException | InterruptException
interruptedException) {
- return;
- }
+ // in state updater we never interrupt the thread
+ // or wakeup the consumer, hence any of
+ // InterruptException / InterruptedException /
WakeupException
+ // should never happen
Review Comment:
We can get rid of this comment since `runOnce()` does not throw a
`InterruptedException` anymore.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -564,14 +566,26 @@ public void start() {
// initialize the last commit as of now to prevent first commit
happens immediately
this.lastCommitMs = time.milliseconds();
+
+ log.info("StateUpdater thread started");
Review Comment:
I would let the state updater thread write this log message, not the thread
that starts the state updater thread.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -312,16 +305,20 @@ private void
addToExceptionsAndFailedTasksThenClearUpdatingTasks(final Exception
updatingTasks.clear();
}
- private void waitIfAllChangelogsCompletelyRead() throws
InterruptedException {
- if (isRunning.get() && changelogReader.allChangelogsCompleted()) {
- tasksAndActionsLock.lock();
- try {
- while (tasksAndActions.isEmpty() &&
!isTopologyResumed.get()) {
- tasksAndActionsCondition.await();
- }
- } finally {
- tasksAndActionsLock.unlock();
+ private void waitIfAllChangelogsCompletelyRead() {
+ tasksAndActionsLock.lock();
Review Comment:
I agree! Actually, I would change it in this PR since it improves the
readability of the code.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -312,16 +305,20 @@ private void
addToExceptionsAndFailedTasksThenClearUpdatingTasks(final Exception
updatingTasks.clear();
}
- private void waitIfAllChangelogsCompletelyRead() throws
InterruptedException {
- if (isRunning.get() && changelogReader.allChangelogsCompleted()) {
- tasksAndActionsLock.lock();
- try {
- while (tasksAndActions.isEmpty() &&
!isTopologyResumed.get()) {
- tasksAndActionsCondition.await();
- }
- } finally {
- tasksAndActionsLock.unlock();
+ private void waitIfAllChangelogsCompletelyRead() {
+ tasksAndActionsLock.lock();
+ try {
+ while (isRunning.get() &&
+ changelogReader.allChangelogsCompleted() &&
+ tasksAndActions.isEmpty() &&
+ !isTopologyResumed.get()) {
+ tasksAndActionsCondition.await();
}
Review Comment:
nit:
```suggestion
while (isRunning.get() &&
changelogReader.allChangelogsCompleted() &&
tasksAndActions.isEmpty() &&
!isTopologyResumed.get()) {
tasksAndActionsCondition.await();
}
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -564,14 +566,26 @@ public void start() {
// initialize the last commit as of now to prevent first commit
happens immediately
this.lastCommitMs = time.milliseconds();
+
+ log.info("StateUpdater thread started");
Review Comment:
I just realized that we have the same log message on line 134. Can we remove
this one? Should we also move the logger back to the thread?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
