You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "cadonna (via GitHub)" <gi...@apache.org> on 2023/08/07 14:38:23 UTC
[GitHub] [kafka] cadonna commented on a diff in pull request #14001: Kafka Streams Threading: Punctuation (5/N)
cadonna commented on code in PR #14001:
URL: https://github.com/apache/kafka/pull/14001#discussion_r1285936281
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java:
##########
@@ -87,11 +88,29 @@ private void runOnce(final long nowMs) {
if (currentTask == null) {
currentTask = taskManager.assignNextTask(DefaultTaskExecutor.this);
} else {
+ boolean progressed = false;
+
+ // First, attempt to punctuate
Review Comment:
nit: I do not think we need this comment. It is clear from the code that we first punctuate.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java:
##########
@@ -87,11 +88,29 @@ private void runOnce(final long nowMs) {
if (currentTask == null) {
currentTask = taskManager.assignNextTask(DefaultTaskExecutor.this);
} else {
+ boolean progressed = false;
+
+ // First, attempt to punctuate
+ if (taskExecutionMetadata.canPunctuateTask(currentTask)) {
+ if (currentTask.maybePunctuateStreamTime()) {
+ log.trace("punctuated stream time for task {} ", currentTask.id());
+ progressed = true;
+ }
+ if (currentTask.maybePunctuateSystemTime()) {
+ log.trace("punctuated system time for task {} ", currentTask.id());
+ progressed = true;
+ }
+ }
Review Comment:
Is there a specific reason we punctuate before we process. In the current code path we first process and then punctuate.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java:
##########
@@ -87,11 +88,29 @@ private void runOnce(final long nowMs) {
if (currentTask == null) {
currentTask = taskManager.assignNextTask(DefaultTaskExecutor.this);
} else {
+ boolean progressed = false;
+
+ // First, attempt to punctuate
+ if (taskExecutionMetadata.canPunctuateTask(currentTask)) {
+ if (currentTask.maybePunctuateStreamTime()) {
+ log.trace("punctuated stream time for task {} ", currentTask.id());
+ progressed = true;
+ }
+ if (currentTask.maybePunctuateSystemTime()) {
+ log.trace("punctuated system time for task {} ", currentTask.id());
+ progressed = true;
+ }
+ }
+
// if a task is no longer processable, ask task-manager to give it another
// task in the next iteration
Review Comment:
This comment does not match the code it should describe. Can we please remove it?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java:
##########
@@ -87,11 +88,29 @@ private void runOnce(final long nowMs) {
if (currentTask == null) {
currentTask = taskManager.assignNextTask(DefaultTaskExecutor.this);
} else {
Review Comment:
Why can we not process a newly assigned task immediately but we wait until the next iteration?
--
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