You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucliu1108 (via GitHub)" <gi...@apache.org> on 2025/10/13 00:08:46 UTC
[PR] KAFKA-19682: Add logs for Kafka Streams task readiness [kafka]
lucliu1108 opened a new pull request, #20693:
URL: https://github.com/apache/kafka/pull/20693
## What
Ticket: https://issues.apache.org/jira/browse/KAFKA-19682
Add more debug/trace logs following the logic of how TaskExecutor process task.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2496663975
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java:
##########
@@ -105,47 +106,49 @@ boolean readyToProcess(final long wallClockTime) {
bufferedPartitions.add(entry.getKey());
}
}
- logger.trace("Ready for processing because max.task.idle.ms is disabled." +
- "\n\tThere may be out-of-order processing for this task as a result." +
- "\n\tBuffered partitions: {}" +
- "\n\tNon-buffered partitions: {}",
- bufferedPartitions,
- emptyPartitions);
+ logMessage = Optional.of(String.format(
+ "Ready for processing because max.task.idle.ms is disabled.%n" +
+ "\tThere may be out-of-order processing for this task as a result.%n" +
+ "\tBuffered partitions: %s%n" +
+ "\tNon-buffered partitions: %s",
+ bufferedPartitions,
+ emptyPartitions
+ ));
}
- return true;
+ return new ReadyToProcessResult(true, logMessage);
}
final Set<TopicPartition> queued = new HashSet<>();
Map<TopicPartition, Long> enforced = null;
+ final StringBuilder logMessageBuilder = new StringBuilder();
for (final Map.Entry<TopicPartition, RecordQueue> entry : partitionQueues.entrySet()) {
final TopicPartition partition = entry.getKey();
final RecordQueue queue = entry.getValue();
-
if (!queue.isEmpty()) {
// this partition is ready for processing
+ appendLog(logMessageBuilder, String.format("Partition %s has buffered data, ready for processing", partition));
Review Comment:
Similar to my commend above: for the positive case (we do return `true` later, we don't really care. We would still log at TRACE directly, but the actually interesting case is the "false" case).
Applies further below, too.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2449521505
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -450,7 +450,7 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean)
}
hasPendingTxCommit = eosEnabled;
- log.debug("Prepared {} task for committing", state());
+ log.debug("Prepared {} task {} for committing", state(), id);
Review Comment:
Yes it is already in the prefix. Have them removed in the message.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501374450
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -1599,6 +1605,70 @@ public void shouldBeRecordIdlingTimeIfSuspended() {
assertThat("task is not idling", task.timeCurrentIdlingStarted().isEmpty());
}
+ @Test
+ public void shouldLogNotReadyWhenStaleAfterThreshold() throws Exception {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+
+ task = createStatelessTask(createConfig("100"));
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 0)));
+
+ // Set lastNotReadyLogTime to simulate staleness
+ final Field lastNotReadyLogTimeField = StreamTask.class.getDeclaredField("lastNotReadyLogTime");
+ lastNotReadyLogTimeField.setAccessible(true);
+
+ try (final LogCaptureAppender streamTaskAppender = LogCaptureAppender.createAndRegister(StreamTask.class);
+ final LogCaptureAppender partitionGroupAppender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) {
+
+ // Enable TRACE logging for PartitionGroup to capture the "ready for processing" message
+ partitionGroupAppender.setClassLogger(PartitionGroup.class, Level.TRACE);
+
+ // Set lastNotReadyLogTime to 100 seconds ago
+ final long initialTime = time.milliseconds();
+ lastNotReadyLogTimeField.set(task, Optional.of(initialTime - 100_000L));
+
+ // Advance time by approximately 5 seconds
+ long newTime = time.milliseconds() + 5_000L;
Review Comment:
If we want to be "exactly, why not use `19_999L` here, ie, exactly 1ms before the cut-off point?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2453330639
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java:
##########
@@ -255,24 +257,32 @@ record = queue.poll(wallClockTime);
if (record != null) {
totalBuffered -= oldSize - queue.size();
+ logger.trace("Partition {} polling next record:, oldSize={}, newSize={}, totalBuffered={}, recordTimestamp={}",
+ queue.partition(), oldSize, queue.size(), totalBuffered, record.timestamp);
if (queue.isEmpty()) {
// if a certain queue has been drained, reset the flag
allBuffered = false;
+ logger.trace("Partition {} queue is now empty, allBuffered=false", queue.partition());
} else {
nonEmptyQueuesByTime.offer(queue);
}
// always update the stream-time to the record's timestamp yet to be processed if it is larger
if (record.timestamp > streamTime) {
+ final long oldStreamTime = streamTime;
streamTime = record.timestamp;
recordLatenessSensor.record(0, wallClockTime);
+ logger.trace("Partition {} stream time updated from {} to {}", queue.partition(), oldStreamTime, streamTime);
} else {
- recordLatenessSensor.record(streamTime - record.timestamp, wallClockTime);
+ final long lateness = streamTime - record.timestamp;
+ recordLatenessSensor.record(lateness, wallClockTime);
}
}
+ } else {
+ logger.trace("Partition pulling nextRecord: no queue available");
}
-
+
Review Comment:
Just some whitespace noise
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2432418610
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -65,9 +67,11 @@ public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
+ log.debug("Task {} has no named topologies", task.id());
return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
} else {
if (pausedTopologies.contains(topologyName)) {
+ log.debug("Task {} has paused topologies", task.id());
return false;
} else {
final NamedTopologyMetadata metadata = topologyNameToErrorMetadata.get(topologyName);
Review Comment:
adding logs here 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.
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2496681941
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -37,20 +37,27 @@
public class TaskExecutionMetadata {
// TODO: implement exponential backoff, for now we just wait 5s
private static final long CONSTANT_BACKOFF_MS = 5_000L;
+ private static final long NOT_READY_LOG_INTERVAL_MS = 120_000L; // Log interval of not being ready (2 minutes)
private final boolean hasNamedTopologies;
private final Set<String> pausedTopologies;
private final ProcessingMode processingMode;
private final Collection<Task> successfullyProcessed = new HashSet<>();
// map of topologies experiencing errors/currently under backoff
private final ConcurrentHashMap<String, NamedTopologyMetadata> topologyNameToErrorMetadata = new ConcurrentHashMap<>();
+ // map of task to last not ready logging time (not present means ready, >= 0 means not ready and tracks last log time)
+ private final ConcurrentHashMap<TaskId, Long> taskToLastNotReadyLogTime = new ConcurrentHashMap<>();
+ // map of task to time when idling started
+ private final ConcurrentHashMap<TaskId, Long> taskToIdlingStartTime = new ConcurrentHashMap<>();
Review Comment:
I don't think we need this idle-timer here -- the difference is, that a user can call `KafkaStreams#pause()`, what we track here.
The idle-time inside a task, has a different purpose: it's for the case that there is more than one input topic partition for a task, and the task cannot make progress because one partitions has no data. For this case, we don't want to block the task forever, and unblock it to process the data we have after "max idle time" passed.
But if a task is `paused()`, we are not "idel" waiting for new input data.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2449519308
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -65,12 +67,15 @@ public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
+ log.debug("Task {} processing check for unnamed topology '{}'", task.id(), topologyName);
Review Comment:
The condition with no `namedTopologies` and the check for named topology task are called often. I downgraded them to TRACE level.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2496674581
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -725,25 +729,49 @@ public boolean isProcessable(final long wallClockTime) {
// a task is only closing / closed when 1) task manager is closing, 2) a rebalance is undergoing;
// in either case we can just log it and move on without notifying the thread since the consumer
// would soon be updated to not return any records for this task anymore.
- log.info("Stream task {} is already in {} state, skip processing it.", id(), state());
+ log.info("Task is already in {} state, skip processing it.", state());
return false;
}
+ final boolean wasReady = lastNotReadyLogTime.isEmpty();
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ if (wasReady) {
+ // READY -> NOT_READY - start timer
+ lastNotReadyLogTime = Optional.of(wallClockTime);
+ } else {
+ maybeLogNotReady(wallClockTime, "Task is not ready to process: has pending transaction commit");
+ }
return false;
}
- final boolean readyToProcess = partitionGroup.readyToProcess(wallClockTime);
- if (!readyToProcess) {
- if (timeCurrentIdlingStarted.isEmpty()) {
+
+ final AbstractPartitionGroup.ReadyToProcessResult readyToProcess = partitionGroup.readyToProcess(wallClockTime);
+ if (!readyToProcess.isReady()) {
+ if (wasReady) {
+ // READY -> NOT_READY - start the timer
+ lastNotReadyLogTime = Optional.of(wallClockTime);
timeCurrentIdlingStarted = Optional.of(wallClockTime);
+ } else {
+ maybeLogNotReady(wallClockTime, readyToProcess.getLogMessage().orElse("Task is not ready to process"));
}
} else {
+ // Task is ready - clear the timer
+ lastNotReadyLogTime = Optional.empty();
timeCurrentIdlingStarted = Optional.empty();
+ log.trace("Task is ready to process");
+ }
+ return readyToProcess.isReady();
+ }
+
+ private void maybeLogNotReady(final long wallClockTime, final String logMessage) {
+ // NOT_READY - check if it should log
+ final long timeSinceLastLog = lastNotReadyLogTime.map(aLong -> wallClockTime - aLong).orElse(0L);
Review Comment:
```suggestion
final long timeSinceLastLog = lastNotReadyLogTime.map(aLong -> wallClockTime - aLong).orElse(-1L);
```
We never logged, so `-1L` is maybe a little be more expressive -- does not really change much; just seems to be a tiny bit cleaner.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501375013
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -1599,6 +1605,70 @@ public void shouldBeRecordIdlingTimeIfSuspended() {
assertThat("task is not idling", task.timeCurrentIdlingStarted().isEmpty());
}
+ @Test
+ public void shouldLogNotReadyWhenStaleAfterThreshold() throws Exception {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+
+ task = createStatelessTask(createConfig("100"));
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 0)));
+
+ // Set lastNotReadyLogTime to simulate staleness
+ final Field lastNotReadyLogTimeField = StreamTask.class.getDeclaredField("lastNotReadyLogTime");
+ lastNotReadyLogTimeField.setAccessible(true);
+
+ try (final LogCaptureAppender streamTaskAppender = LogCaptureAppender.createAndRegister(StreamTask.class);
+ final LogCaptureAppender partitionGroupAppender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) {
+
+ // Enable TRACE logging for PartitionGroup to capture the "ready for processing" message
+ partitionGroupAppender.setClassLogger(PartitionGroup.class, Level.TRACE);
+
+ // Set lastNotReadyLogTime to 100 seconds ago
+ final long initialTime = time.milliseconds();
+ lastNotReadyLogTimeField.set(task, Optional.of(initialTime - 100_000L));
+
+ // Advance time by approximately 5 seconds
Review Comment:
Approximately? Seems be be exactly 5 seconds :)
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501616928
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -1599,6 +1605,70 @@ public void shouldBeRecordIdlingTimeIfSuspended() {
assertThat("task is not idling", task.timeCurrentIdlingStarted().isEmpty());
}
+ @Test
+ public void shouldLogNotReadyWhenStaleAfterThreshold() throws Exception {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+
+ task = createStatelessTask(createConfig("100"));
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 0)));
+
+ // Set lastNotReadyLogTime to simulate staleness
+ final Field lastNotReadyLogTimeField = StreamTask.class.getDeclaredField("lastNotReadyLogTime");
+ lastNotReadyLogTimeField.setAccessible(true);
+
+ try (final LogCaptureAppender streamTaskAppender = LogCaptureAppender.createAndRegister(StreamTask.class);
+ final LogCaptureAppender partitionGroupAppender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) {
+
+ // Enable TRACE logging for PartitionGroup to capture the "ready for processing" message
+ partitionGroupAppender.setClassLogger(PartitionGroup.class, Level.TRACE);
+
+ // Set lastNotReadyLogTime to 100 seconds ago
+ final long initialTime = time.milliseconds();
+ lastNotReadyLogTimeField.set(task, Optional.of(initialTime - 100_000L));
+
+ // Advance time by approximately 5 seconds
+ long newTime = time.milliseconds() + 5_000L;
Review Comment:
If we use a time passed very close to the threshold, it might cause some flakiness (although of very low probability) that initial time and `time.milliseconds()` pass 1ms.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2432365087
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -788,9 +795,16 @@ record = partitionGroup.nextRecord(recordInfo, wallClockTime);
consumedOffsets.put(partition, record.offset());
commitNeeded = true;
+ if (log.isTraceEnabled()) {
+ log.trace("Task {} processed record: topic={}, partition={}, offset={}",
+ id, record.topic(), record.partition(), record.offset());
+ }
+
// after processing this record, if its partition queue's buffered size has been
// decreased to the threshold, we can then resume the consumption on this partition
if (recordInfo.queue().size() <= maxBufferedSize) {
+ log.trace("Resume consumption for partition {}: buffered size {} is under the threshold {}",
+ partition, recordInfo.queue().size(), maxBufferedSize);
partitionsToResume.add(partition);
}
Review Comment:
Does that make sense to add logs for catch-clauses 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.
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501350396
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -63,16 +68,52 @@ public ProcessingMode processingMode() {
public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
+ final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
+ final boolean canProcess;
+ final String logMessage;
+
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
- return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
+ logMessage = String.format("Task %s processing check for unnamed topology '%s'", task.id(), topologyName);
Review Comment:
```suggestion
logMessage = String.format("Task %s processing check", task.id();
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -63,16 +68,52 @@ public ProcessingMode processingMode() {
public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
+ final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
+ final boolean canProcess;
+ final String logMessage;
+
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
- return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
+ logMessage = String.format("Task %s processing check for unnamed topology '%s'", task.id(), topologyName);
Review Comment:
```suggestion
logMessage = String.format("Task %s processing check", task.id());
```
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501364694
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -63,16 +68,52 @@ public ProcessingMode processingMode() {
public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
+ final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
+ final boolean canProcess;
+ final String logMessage;
+
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
- return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
+ logMessage = String.format("Task %s processing check for unnamed topology '%s'", task.id(), topologyName);
+ canProcess = !pausedTopologies.contains(UNNAMED_TOPOLOGY);
} else {
if (pausedTopologies.contains(topologyName)) {
- return false;
+ canProcess = false;
+ logMessage = String.format("Task %s can't be processed: topology '%s' is paused", task.id(), topologyName);
} else {
+ logMessage = String.format("Task %s processing check for named topology '%s'", task.id(), topologyName);
Review Comment:
As above. Should we move this after `canProcess = metadata == null || (metadata.canProcess() && metadata.canProcessTask(task, now));` and have a different message depending if `canProcess` is true vs false?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501356226
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -63,16 +68,52 @@ public ProcessingMode processingMode() {
public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
+ final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
+ final boolean canProcess;
+ final String logMessage;
+
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
- return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
+ logMessage = String.format("Task %s processing check for unnamed topology '%s'", task.id(), topologyName);
Review Comment:
But I am wondering if we would want to just use `Task %s can't be processed: topology is paused` -- similar to the "name topology code below?
In the end, we are preparing an error message, which we only use if we are not "ready" for processing, so we only need the "negative" one?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501453430
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -1599,6 +1605,70 @@ public void shouldBeRecordIdlingTimeIfSuspended() {
assertThat("task is not idling", task.timeCurrentIdlingStarted().isEmpty());
}
+ @Test
+ public void shouldLogNotReadyWhenStaleAfterThreshold() throws Exception {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+
+ task = createStatelessTask(createConfig("100"));
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 0)));
+
+ // Set lastNotReadyLogTime to simulate staleness
+ final Field lastNotReadyLogTimeField = StreamTask.class.getDeclaredField("lastNotReadyLogTime");
+ lastNotReadyLogTimeField.setAccessible(true);
+
+ try (final LogCaptureAppender streamTaskAppender = LogCaptureAppender.createAndRegister(StreamTask.class);
+ final LogCaptureAppender partitionGroupAppender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) {
+
+ // Enable TRACE logging for PartitionGroup to capture the "ready for processing" message
+ partitionGroupAppender.setClassLogger(PartitionGroup.class, Level.TRACE);
+
+ // Set lastNotReadyLogTime to 100 seconds ago
+ final long initialTime = time.milliseconds();
+ lastNotReadyLogTimeField.set(task, Optional.of(initialTime - 100_000L));
+
+ // Advance time by approximately 5 seconds
Review Comment:
i'm thinking about the time elapsed between initialTime to time.milliseconds() . should be a little bit more than 5 seconds (but that difference is trivial)
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501356226
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -63,16 +68,52 @@ public ProcessingMode processingMode() {
public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
+ final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
+ final boolean canProcess;
+ final String logMessage;
+
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
- return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
+ logMessage = String.format("Task %s processing check for unnamed topology '%s'", task.id(), topologyName);
Review Comment:
But I am wondering if we would want to just use `Task %s can't be processed: topology is paused` -- similar to the "name topology code below?
In the end, we are preparing an error message, which we only use if we are not "ready" for processing, so we only need the "negative" one?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -63,16 +68,52 @@ public ProcessingMode processingMode() {
public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
+ final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
+ final boolean canProcess;
+ final String logMessage;
+
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
- return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
+ logMessage = String.format("Task %s processing check for unnamed topology '%s'", task.id(), topologyName);
+ canProcess = !pausedTopologies.contains(UNNAMED_TOPOLOGY);
} else {
if (pausedTopologies.contains(topologyName)) {
- return false;
+ canProcess = false;
+ logMessage = String.format("Task %s can't be processed: topology '%s' is paused", task.id(), topologyName);
} else {
+ logMessage = String.format("Task %s processing check for named topology '%s'", task.id(), topologyName);
Review Comment:
Not sure if we need this error message? At this point, we don' know yet, if we can or cannot process, as we compute this only below via `canProcess = metadata == null || (metadata.canProcess() && metadata.canProcessTask(task, now));`.
But we would use the error message only if `canProcess` is set to `false`, so should we not just reuse ` logMessage = String.format("Task %s can't be processed: topology '%s' is paused", task.id(), topologyName);` from above?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501379517
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java:
##########
@@ -112,40 +112,40 @@ boolean readyToProcess(final long wallClockTime) {
bufferedPartitions,
emptyPartitions);
}
- return true;
+ return new ReadyToProcessResult(true, Optional.empty());
}
final Set<TopicPartition> queued = new HashSet<>();
Map<TopicPartition, Long> enforced = null;
+ final StringBuilder logMessageBuilder = new StringBuilder();
for (final Map.Entry<TopicPartition, RecordQueue> entry : partitionQueues.entrySet()) {
final TopicPartition partition = entry.getKey();
final RecordQueue queue = entry.getValue();
-
if (!queue.isEmpty()) {
// this partition is ready for processing
+ logger.trace("Partition {} has buffered data, ready for processing", partition);
idlePartitionDeadlines.remove(partition);
queued.add(partition);
} else {
final Long fetchedLag = fetchedLags.getOrDefault(partition, -1L);
-
- logger.trace("Fetched lag for {} is {}", partition, fetchedLag);
-
+ appendLog(logMessageBuilder, String.format("Partition %s has fetched lag for %d", partition, fetchedLag));
Review Comment:
```suggestion
appendLog(logMessageBuilder, String.format("Partition %s has fetched lag of %d", partition, fetchedLag));
```
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2492145577
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -725,27 +727,79 @@ public boolean isProcessable(final long wallClockTime) {
// a task is only closing / closed when 1) task manager is closing, 2) a rebalance is undergoing;
// in either case we can just log it and move on without notifying the thread since the consumer
// would soon be updated to not return any records for this task anymore.
- log.info("Stream task {} is already in {} state, skip processing it.", id(), state());
+ log.info("Task is already in {} state, skip processing it.", state());
return false;
}
+ final boolean wasReady = lastNotReadyLogTime == -1L;
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ if (wasReady) {
+ // READY -> NOT_READY - start timer
+ lastNotReadyLogTime = wallClockTime;
+ } else {
+ // NOT_READY - check if it should log
+ final long timeSinceLastLog = wallClockTime - lastNotReadyLogTime;
+ if (timeSinceLastLog >= NOT_READY_LOG_INTERVAL_MS) {
+ log.info("Task is not ready to process: has pending transaction commit");
+ lastNotReadyLogTime = wallClockTime;
+ }
+ }
return false;
}
+
final boolean readyToProcess = partitionGroup.readyToProcess(wallClockTime);
if (!readyToProcess) {
- if (timeCurrentIdlingStarted.isEmpty()) {
+ if (wasReady) {
+ // READY -> NOT_READY - start the timer
+ lastNotReadyLogTime = wallClockTime;
timeCurrentIdlingStarted = Optional.of(wallClockTime);
+ } else {
+ // NOT_READY - check if it should log
+ final long timeSinceLastLog = wallClockTime - lastNotReadyLogTime;
+ if (timeSinceLastLog >= NOT_READY_LOG_INTERVAL_MS) {
+ final String partitionStatus = getNotReadyPartitionStatus();
+ log.info("Task is not ready to process (started idling at time {}): {}", timeCurrentIdlingStarted.orElse(-1L), partitionStatus);
Review Comment:
It seems this message is rather generic -- guess it's the question you asked? -- I think it would be good to modify `partitionGroup.readyToProcess(...)` to allow us to get a better error message.
I see three ways:
1. change the return type to `String` (or `Optional<String>`) and use `null`/empty to indicate "read" -- not super clean but would work
2. add a second parameter `readyToProcess(long wallClockTime, Optional<String> returnedLogMessage)` (or similar), which we set before `return false` -- also not ideal to use an input parameter but again, would work.
3. add a new helper class that wraps `boolean readyToProcess` and `Optional<String> logMessage` and use a return type of `readyToProcess` (this is technically the cleanest way, but most involved change)
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -725,27 +727,79 @@ public boolean isProcessable(final long wallClockTime) {
// a task is only closing / closed when 1) task manager is closing, 2) a rebalance is undergoing;
// in either case we can just log it and move on without notifying the thread since the consumer
// would soon be updated to not return any records for this task anymore.
- log.info("Stream task {} is already in {} state, skip processing it.", id(), state());
+ log.info("Task is already in {} state, skip processing it.", state());
return false;
}
+ final boolean wasReady = lastNotReadyLogTime == -1L;
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ if (wasReady) {
+ // READY -> NOT_READY - start timer
+ lastNotReadyLogTime = wallClockTime;
+ } else {
+ // NOT_READY - check if it should log
+ final long timeSinceLastLog = wallClockTime - lastNotReadyLogTime;
+ if (timeSinceLastLog >= NOT_READY_LOG_INTERVAL_MS) {
+ log.info("Task is not ready to process: has pending transaction commit");
+ lastNotReadyLogTime = wallClockTime;
+ }
+ }
return false;
}
+
final boolean readyToProcess = partitionGroup.readyToProcess(wallClockTime);
if (!readyToProcess) {
- if (timeCurrentIdlingStarted.isEmpty()) {
+ if (wasReady) {
+ // READY -> NOT_READY - start the timer
+ lastNotReadyLogTime = wallClockTime;
timeCurrentIdlingStarted = Optional.of(wallClockTime);
+ } else {
+ // NOT_READY - check if it should log
+ final long timeSinceLastLog = wallClockTime - lastNotReadyLogTime;
+ if (timeSinceLastLog >= NOT_READY_LOG_INTERVAL_MS) {
+ final String partitionStatus = getNotReadyPartitionStatus();
Review Comment:
If we really get the details from `partitionGroup.readyToProcess`, it seems we won't seen this?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -725,27 +727,79 @@ public boolean isProcessable(final long wallClockTime) {
// a task is only closing / closed when 1) task manager is closing, 2) a rebalance is undergoing;
// in either case we can just log it and move on without notifying the thread since the consumer
// would soon be updated to not return any records for this task anymore.
- log.info("Stream task {} is already in {} state, skip processing it.", id(), state());
+ log.info("Task is already in {} state, skip processing it.", state());
return false;
}
+ final boolean wasReady = lastNotReadyLogTime == -1L;
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ if (wasReady) {
+ // READY -> NOT_READY - start timer
+ lastNotReadyLogTime = wallClockTime;
+ } else {
+ // NOT_READY - check if it should log
+ final long timeSinceLastLog = wallClockTime - lastNotReadyLogTime;
+ if (timeSinceLastLog >= NOT_READY_LOG_INTERVAL_MS) {
+ log.info("Task is not ready to process: has pending transaction commit");
+ lastNotReadyLogTime = wallClockTime;
+ }
Review Comment:
It seems we can extract this into a helper method (`maybeLogNotReady(...)`, just passing in the log message? This code is always the same:
```
final long timeSinceLastLog = wallClockTime - lastNotReadyLogTime;
if (timeSinceLastLog >= NOT_READY_LOG_INTERVAL_MS) {
log.info(logMessage);
lastNotReadyLogTime = wallClockTime;
}
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -118,7 +118,9 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
private boolean commitNeeded = false;
private boolean commitRequested = false;
private boolean hasPendingTxCommit = false;
- private Optional<Long> timeCurrentIdlingStarted;
+ private Optional<Long> timeCurrentIdlingStarted; // time since the task started idling, empty if not idling
+ private long lastNotReadyLogTime = -1L; // -1 means ready (no logging needed), >= 0 means not ready and tracks last log time
Review Comment:
Nit: we use `Optional<Long>` for `timeCurrentIdlingStarted` -- should we do the same for `lastNotReadyLogTime` to avoid using two different pattern?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -65,13 +67,17 @@ public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
+ log.trace("Task {} processing check for unnamed topology '{}'", task.id(), topologyName);
Review Comment:
I think we want to include this in the periodic logging, too? (same for the the new logs below?)
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2496661085
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java:
##########
@@ -93,9 +93,10 @@ class PartitionGroup extends AbstractPartitionGroup {
}
@Override
- boolean readyToProcess(final long wallClockTime) {
+ ReadyToProcessResult readyToProcess(final long wallClockTime) {
+ Optional<String> logMessage = Optional.empty();
if (maxTaskIdleMs == StreamsConfig.MAX_TASK_IDLE_MS_DISABLED) {
- if (logger.isTraceEnabled() && !allBuffered && totalBuffered > 0) {
+ if (!allBuffered && totalBuffered > 0) {
Review Comment:
I believe we would want to keep this unchanged, and just return `new ReadyToProcessResult(true, Optional.empty());`
The existing trace log is for the case when we are processable, and the idea of `ReadyToProcessResult` is return a log message only for the "not processable" case.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2449525658
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -733,13 +733,15 @@ public boolean isProcessable(final long wallClockTime) {
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ log.debug("Stream task {} has a pending transaction commit, skip processing it.", id());
return false;
}
final boolean readyToProcess = partitionGroup.readyToProcess(wallClockTime);
if (!readyToProcess) {
if (timeCurrentIdlingStarted.isEmpty()) {
timeCurrentIdlingStarted = Optional.of(wallClockTime);
}
+ log.debug("Task {} started idling at time {}", id, timeCurrentIdlingStarted.get());
} else {
timeCurrentIdlingStarted = Optional.empty();
Review Comment:
Added .Since this branch is called more often I set it to TRACE
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on PR #20693:
URL: https://github.com/apache/kafka/pull/20693#issuecomment-3488042235
@mjsax I just added periodic logging for `StreamTask#isProcessable`.
For the NOT_READY reason, do we need to include the logging message from the inner `PartitionGroup#readyToProcess` as part for the `isProcessable()` method if it stays in NOT_READY state (which might require some changes to the original PartitionGroup interface)? Or just the partition status is enough message?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2496663975
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java:
##########
@@ -105,47 +106,49 @@ boolean readyToProcess(final long wallClockTime) {
bufferedPartitions.add(entry.getKey());
}
}
- logger.trace("Ready for processing because max.task.idle.ms is disabled." +
- "\n\tThere may be out-of-order processing for this task as a result." +
- "\n\tBuffered partitions: {}" +
- "\n\tNon-buffered partitions: {}",
- bufferedPartitions,
- emptyPartitions);
+ logMessage = Optional.of(String.format(
+ "Ready for processing because max.task.idle.ms is disabled.%n" +
+ "\tThere may be out-of-order processing for this task as a result.%n" +
+ "\tBuffered partitions: %s%n" +
+ "\tNon-buffered partitions: %s",
+ bufferedPartitions,
+ emptyPartitions
+ ));
}
- return true;
+ return new ReadyToProcessResult(true, logMessage);
}
final Set<TopicPartition> queued = new HashSet<>();
Map<TopicPartition, Long> enforced = null;
+ final StringBuilder logMessageBuilder = new StringBuilder();
for (final Map.Entry<TopicPartition, RecordQueue> entry : partitionQueues.entrySet()) {
final TopicPartition partition = entry.getKey();
final RecordQueue queue = entry.getValue();
-
if (!queue.isEmpty()) {
// this partition is ready for processing
+ appendLog(logMessageBuilder, String.format("Partition %s has buffered data, ready for processing", partition));
Review Comment:
Similar to my commend above: for the positive case (we do return `true` later, we don't really care. We would still log at TRACE directly, but the actually interesting case is the "false" case).
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501375472
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -1599,6 +1605,70 @@ public void shouldBeRecordIdlingTimeIfSuspended() {
assertThat("task is not idling", task.timeCurrentIdlingStarted().isEmpty());
}
+ @Test
+ public void shouldLogNotReadyWhenStaleAfterThreshold() throws Exception {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+
+ task = createStatelessTask(createConfig("100"));
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 0)));
+
+ // Set lastNotReadyLogTime to simulate staleness
+ final Field lastNotReadyLogTimeField = StreamTask.class.getDeclaredField("lastNotReadyLogTime");
+ lastNotReadyLogTimeField.setAccessible(true);
+
+ try (final LogCaptureAppender streamTaskAppender = LogCaptureAppender.createAndRegister(StreamTask.class);
+ final LogCaptureAppender partitionGroupAppender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) {
+
+ // Enable TRACE logging for PartitionGroup to capture the "ready for processing" message
+ partitionGroupAppender.setClassLogger(PartitionGroup.class, Level.TRACE);
+
+ // Set lastNotReadyLogTime to 100 seconds ago
+ final long initialTime = time.milliseconds();
+ lastNotReadyLogTimeField.set(task, Optional.of(initialTime - 100_000L));
+
+ // Advance time by approximately 5 seconds
+ long newTime = time.milliseconds() + 5_000L;
+
+ // Should not trigger logging after being stale for 105 seconds
+ assertFalse(task.isProcessable(newTime));
+ List<String> messages = streamTaskAppender.getMessages();
+ assertEquals(0, messages.size(), "No log message should be logged before 20 seconds");
+
+ // Advance time by approximately 20 seconds
Review Comment:
Approximately?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501360875
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -63,16 +68,52 @@ public ProcessingMode processingMode() {
public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
+ final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
+ final boolean canProcess;
+ final String logMessage;
+
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
- return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
+ logMessage = String.format("Task %s processing check for unnamed topology '%s'", task.id(), topologyName);
+ canProcess = !pausedTopologies.contains(UNNAMED_TOPOLOGY);
} else {
if (pausedTopologies.contains(topologyName)) {
- return false;
+ canProcess = false;
+ logMessage = String.format("Task %s can't be processed: topology '%s' is paused", task.id(), topologyName);
} else {
+ logMessage = String.format("Task %s processing check for named topology '%s'", task.id(), topologyName);
Review Comment:
Not sure if we need this error message? At this point, we don' know yet, if we can or cannot process, as we compute this only below via `canProcess = metadata == null || (metadata.canProcess() && metadata.canProcessTask(task, now));`.
But we would use the error message only if `canProcess` is set to `false`, so should we not just reuse ` logMessage = String.format("Task %s can't be processed: topology '%s' is paused", task.id(), topologyName);` from above?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2505399958
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -1625,15 +1625,15 @@ public void shouldLogNotReadyWhenStaleAfterThreshold() throws Exception {
final long initialTime = time.milliseconds();
task.setLastNotReadyLogTime(initialTime - 100_000L);
- // Advance time by approximately 5 seconds
- long newTime = time.milliseconds() + 5_000L;
+ // Advance time by 19 seconds
+ long newTime = time.milliseconds() + 19_000L;
Review Comment:
Seems we could make this `19_999` to have a most tight test, which verifies the exact ms cut of point?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2449380906
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java:
##########
@@ -255,24 +257,32 @@ record = queue.poll(wallClockTime);
if (record != null) {
Review Comment:
Record should not be `null`. If it is null then the queue itself is also empty (which is detected in the outer loop).
I tried to add a else statement that throws an exception, and is detected by `SpotBugs` as "redundant null check". Therefore i think it's fine to drop 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.
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2453326964
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -65,13 +67,17 @@ public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
+ log.trace("Task {} processing check for unnamed topology '{}'", task.id(), topologyName);
return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
} else {
if (pausedTopologies.contains(topologyName)) {
+ log.debug("Task {} can't be processed: topology '{}' is paused", task.id(), topologyName);
Review Comment:
```suggestion
log.trace("Task {} can't be processed: topology '{}' is paused", task.id(), topologyName);
```
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2453330075
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java:
##########
@@ -255,24 +257,32 @@ record = queue.poll(wallClockTime);
if (record != null) {
totalBuffered -= oldSize - queue.size();
+ logger.trace("Partition {} polling next record:, oldSize={}, newSize={}, totalBuffered={}, recordTimestamp={}",
+ queue.partition(), oldSize, queue.size(), totalBuffered, record.timestamp);
if (queue.isEmpty()) {
// if a certain queue has been drained, reset the flag
allBuffered = false;
+ logger.trace("Partition {} queue is now empty, allBuffered=false", queue.partition());
} else {
nonEmptyQueuesByTime.offer(queue);
}
// always update the stream-time to the record's timestamp yet to be processed if it is larger
if (record.timestamp > streamTime) {
+ final long oldStreamTime = streamTime;
streamTime = record.timestamp;
recordLatenessSensor.record(0, wallClockTime);
+ logger.trace("Partition {} stream time updated from {} to {}", queue.partition(), oldStreamTime, streamTime);
} else {
- recordLatenessSensor.record(streamTime - record.timestamp, wallClockTime);
+ final long lateness = streamTime - record.timestamp;
+ recordLatenessSensor.record(lateness, wallClockTime);
}
}
+ } else {
+ logger.trace("Partition pulling nextRecord: no queue available");
}
-
+
Review Comment:
```suggestion
```
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2449532669
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java:
##########
@@ -255,24 +257,32 @@ record = queue.poll(wallClockTime);
if (record != null) {
totalBuffered -= oldSize - queue.size();
+ logger.trace("Partition {} polling next record:, oldSize={}, newSize={}, totalBuffered={}, recordTimestamp={}",
+ queue.partition(), oldSize, queue.size(), totalBuffered, record.timestamp);
if (queue.isEmpty()) {
// if a certain queue has been drained, reset the flag
allBuffered = false;
+ logger.trace("Partition {} queue is now empty, allBuffered=false", queue.partition());
} else {
nonEmptyQueuesByTime.offer(queue);
}
// always update the stream-time to the record's timestamp yet to be processed if it is larger
if (record.timestamp > streamTime) {
+ final long oldStreamTime = streamTime;
streamTime = record.timestamp;
recordLatenessSensor.record(0, wallClockTime);
+ logger.trace("Partition {} stream time updated from {} to {}", queue.partition(), oldStreamTime, streamTime);
} else {
- recordLatenessSensor.record(streamTime - record.timestamp, wallClockTime);
+ final long lateness = streamTime - record.timestamp;
+ recordLatenessSensor.record(lateness, wallClockTime);
}
}
+ } else {
+ logger.trace("Partition pulling nextRecord: no queue available, totalBuffered={}", totalBuffered);
Review Comment:
Yes it is 0. This param is removed from the message.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2449564988
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -44,13 +44,15 @@ public class TaskExecutionMetadata {
private final Collection<Task> successfullyProcessed = new HashSet<>();
// map of topologies experiencing errors/currently under backoff
private final ConcurrentHashMap<String, NamedTopologyMetadata> topologyNameToErrorMetadata = new ConcurrentHashMap<>();
+ private final Logger log;
public TaskExecutionMetadata(final Set<String> allTopologyNames,
final Set<String> pausedTopologies,
final ProcessingMode processingMode) {
this.hasNamedTopologies = !(allTopologyNames.size() == 1 && allTopologyNames.contains(UNNAMED_TOPOLOGY));
this.pausedTopologies = pausedTopologies;
this.processingMode = processingMode;
+ this.log = new LogContext("[TaskExecutionMetadata]").logger(TaskExecutionMetadata.class);
Review Comment:
Revised the LogContext to include empty prefix.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2496862620
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -725,25 +729,49 @@ public boolean isProcessable(final long wallClockTime) {
// a task is only closing / closed when 1) task manager is closing, 2) a rebalance is undergoing;
// in either case we can just log it and move on without notifying the thread since the consumer
// would soon be updated to not return any records for this task anymore.
- log.info("Stream task {} is already in {} state, skip processing it.", id(), state());
+ log.info("Task is already in {} state, skip processing it.", state());
return false;
}
+ final boolean wasReady = lastNotReadyLogTime.isEmpty();
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ if (wasReady) {
+ // READY -> NOT_READY - start timer
+ lastNotReadyLogTime = Optional.of(wallClockTime);
+ } else {
+ maybeLogNotReady(wallClockTime, "Task is not ready to process: has pending transaction commit");
+ }
return false;
}
- final boolean readyToProcess = partitionGroup.readyToProcess(wallClockTime);
- if (!readyToProcess) {
- if (timeCurrentIdlingStarted.isEmpty()) {
+
+ final AbstractPartitionGroup.ReadyToProcessResult readyToProcess = partitionGroup.readyToProcess(wallClockTime);
+ if (!readyToProcess.isReady()) {
+ if (wasReady) {
+ // READY -> NOT_READY - start the timer
+ lastNotReadyLogTime = Optional.of(wallClockTime);
timeCurrentIdlingStarted = Optional.of(wallClockTime);
+ } else {
+ maybeLogNotReady(wallClockTime, readyToProcess.getLogMessage().orElse("Task is not ready to process"));
Review Comment:
Yes, if the code is correct, it shouldn't go to this `orElse` condition. Will use `orElseThrow` instead.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2505393570
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -242,6 +242,10 @@ private Map<TopicPartition, RecordQueue> createPartitionQueues() {
return partitionQueues;
}
+ void setLastNotReadyLogTime(final long lastNotReadyLogTime) {
Review Comment:
Let add a comment `// for testing only`
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501363713
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -63,16 +68,52 @@ public ProcessingMode processingMode() {
public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
+ final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
+ final boolean canProcess;
+ final String logMessage;
+
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
- return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
+ logMessage = String.format("Task %s processing check for unnamed topology '%s'", task.id(), topologyName);
+ canProcess = !pausedTopologies.contains(UNNAMED_TOPOLOGY);
Review Comment:
Wondering if we should do this line first, and compute the log message based on the output of `canProcess` ?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2432279248
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -742,13 +742,19 @@ public boolean isProcessable(final long wallClockTime) {
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ if (log.isDebugEnabled()) {
Review Comment:
What is the reason for calling `isDebugEnabled()` only at this specific location and not in the former debug leveling loggings?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -742,13 +742,19 @@ public boolean isProcessable(final long wallClockTime) {
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ if (log.isDebugEnabled()) {
Review Comment:
What is the reason for calling `isDebugEnabled()` only at this specific location and not in the former debug level loggings?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2496670474
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -725,25 +729,49 @@ public boolean isProcessable(final long wallClockTime) {
// a task is only closing / closed when 1) task manager is closing, 2) a rebalance is undergoing;
// in either case we can just log it and move on without notifying the thread since the consumer
// would soon be updated to not return any records for this task anymore.
- log.info("Stream task {} is already in {} state, skip processing it.", id(), state());
+ log.info("Task is already in {} state, skip processing it.", state());
return false;
}
+ final boolean wasReady = lastNotReadyLogTime.isEmpty();
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ if (wasReady) {
+ // READY -> NOT_READY - start timer
+ lastNotReadyLogTime = Optional.of(wallClockTime);
+ } else {
+ maybeLogNotReady(wallClockTime, "Task is not ready to process: has pending transaction commit");
+ }
return false;
}
- final boolean readyToProcess = partitionGroup.readyToProcess(wallClockTime);
- if (!readyToProcess) {
- if (timeCurrentIdlingStarted.isEmpty()) {
+
+ final AbstractPartitionGroup.ReadyToProcessResult readyToProcess = partitionGroup.readyToProcess(wallClockTime);
+ if (!readyToProcess.isReady()) {
+ if (wasReady) {
+ // READY -> NOT_READY - start the timer
+ lastNotReadyLogTime = Optional.of(wallClockTime);
timeCurrentIdlingStarted = Optional.of(wallClockTime);
+ } else {
+ maybeLogNotReady(wallClockTime, readyToProcess.getLogMessage().orElse("Task is not ready to process"));
Review Comment:
For this case, we should always have a log message, right? Wondering if we should use `orElseThrow` to surface a potential bug directly?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -725,25 +729,49 @@ public boolean isProcessable(final long wallClockTime) {
// a task is only closing / closed when 1) task manager is closing, 2) a rebalance is undergoing;
// in either case we can just log it and move on without notifying the thread since the consumer
// would soon be updated to not return any records for this task anymore.
- log.info("Stream task {} is already in {} state, skip processing it.", id(), state());
+ log.info("Task is already in {} state, skip processing it.", state());
return false;
}
+ final boolean wasReady = lastNotReadyLogTime.isEmpty();
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ if (wasReady) {
+ // READY -> NOT_READY - start timer
+ lastNotReadyLogTime = Optional.of(wallClockTime);
+ } else {
+ maybeLogNotReady(wallClockTime, "Task is not ready to process: has pending transaction commit");
+ }
return false;
}
- final boolean readyToProcess = partitionGroup.readyToProcess(wallClockTime);
- if (!readyToProcess) {
- if (timeCurrentIdlingStarted.isEmpty()) {
+
+ final AbstractPartitionGroup.ReadyToProcessResult readyToProcess = partitionGroup.readyToProcess(wallClockTime);
+ if (!readyToProcess.isReady()) {
+ if (wasReady) {
+ // READY -> NOT_READY - start the timer
+ lastNotReadyLogTime = Optional.of(wallClockTime);
timeCurrentIdlingStarted = Optional.of(wallClockTime);
+ } else {
+ maybeLogNotReady(wallClockTime, readyToProcess.getLogMessage().orElse("Task is not ready to process"));
Review Comment:
For this case, we should always have a log message, right? Wondering if we should use `orElseThrow` to surface a potential bug in our code directly?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2496676266
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractPartitionGroup.java:
##########
@@ -25,7 +25,29 @@
abstract class AbstractPartitionGroup {
- abstract boolean readyToProcess(long wallClockTime);
+ /**
+ * Result of readyToProcess check, containing both the readiness status
+ * and an optional diagnostic log message.
+ */
+ static final class ReadyToProcessResult {
+ private final boolean ready;
+ private final Optional<String> logMessage;
+
+ ReadyToProcessResult(final boolean ready, final Optional<String> logMessage) {
+ this.ready = ready;
+ this.logMessage = logMessage;
Review Comment:
Should we add a check, if `ready == true`, `logMessage` cannot be empty, and if `ready == true`, `logMessage` should be empty?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501351695
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -63,16 +68,52 @@ public ProcessingMode processingMode() {
public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
+ final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
+ final boolean canProcess;
+ final String logMessage;
+
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
- return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
+ logMessage = String.format("Task %s processing check for unnamed topology '%s'", task.id(), topologyName);
Review Comment:
If `hasNamedTopologies` is false, `topologyName == null` -- we use `UNNAMED_TOPOLOGY` as in internally helper value only. -- For this case, we should not talk about "named topology" at all.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2501370694
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -1599,6 +1605,70 @@ public void shouldBeRecordIdlingTimeIfSuspended() {
assertThat("task is not idling", task.timeCurrentIdlingStarted().isEmpty());
}
+ @Test
+ public void shouldLogNotReadyWhenStaleAfterThreshold() throws Exception {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+
+ task = createStatelessTask(createConfig("100"));
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 0)));
+
+ // Set lastNotReadyLogTime to simulate staleness
+ final Field lastNotReadyLogTimeField = StreamTask.class.getDeclaredField("lastNotReadyLogTime");
Review Comment:
Wondering if we should better add a package-private method to `StreamTask` allowing us to just set it? Using reflections is always ugly.
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2432352945
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -773,6 +779,7 @@ record = partitionGroup.nextRecord(recordInfo, wallClockTime);
// if there is no record to process, return immediately
if (record == null) {
+ log.trace("Task {} has no next record to process.", id());
return false;
}
}
Review Comment:
Add `else` to the if-clause ` if (!(record instanceof CorruptedRecord)) {` and logs for the else-clause?
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #20693:
URL: https://github.com/apache/kafka/pull/20693#issuecomment-3406070969
Thank you, @lucliu1108, for the PR. A bit tricky to choose between using debug and trace ;-)
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #20693:
URL: https://github.com/apache/kafka/pull/20693#issuecomment-3406236512
Thank you, @lucliu1108, for the PR. A bit tricky to choose between using debug and trace ;-)
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #20693:
URL: https://github.com/apache/kafka/pull/20693
--
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-19682: Add logs for Kafka Streams task readiness [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2445957003
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -44,13 +44,15 @@ public class TaskExecutionMetadata {
private final Collection<Task> successfullyProcessed = new HashSet<>();
// map of topologies experiencing errors/currently under backoff
private final ConcurrentHashMap<String, NamedTopologyMetadata> topologyNameToErrorMetadata = new ConcurrentHashMap<>();
+ private final Logger log;
public TaskExecutionMetadata(final Set<String> allTopologyNames,
final Set<String> pausedTopologies,
final ProcessingMode processingMode) {
this.hasNamedTopologies = !(allTopologyNames.size() == 1 && allTopologyNames.contains(UNNAMED_TOPOLOGY));
this.pausedTopologies = pausedTopologies;
this.processingMode = processingMode;
+ this.log = new LogContext("[TaskExecutionMetadata]").logger(TaskExecutionMetadata.class);
Review Comment:
Why do we pass the class name as log prefix, given that we set the class via `logger(...)` anyway?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java:
##########
@@ -65,12 +67,15 @@ public boolean canProcessTask(final Task task, final long now) {
final String topologyName = task.id().topologyName();
if (!hasNamedTopologies) {
// TODO implement error handling/backoff for non-named topologies (needs KIP)
+ log.debug("Task {} processing check for unnamed topology '{}'", task.id(), topologyName);
Review Comment:
Did we run with DEBUG enabled? This one might become very chatty?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java:
##########
@@ -255,24 +257,32 @@ record = queue.poll(wallClockTime);
if (record != null) {
Review Comment:
Wondering if record be every be `null` here? -- Should we actually add an `else` and log an ERROR (or WARN) log, or even throw `IllegalStateException` as this should always be true?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java:
##########
@@ -255,24 +257,32 @@ record = queue.poll(wallClockTime);
if (record != null) {
totalBuffered -= oldSize - queue.size();
+ logger.trace("Partition {} polling next record:, oldSize={}, newSize={}, totalBuffered={}, recordTimestamp={}",
+ queue.partition(), oldSize, queue.size(), totalBuffered, record.timestamp);
if (queue.isEmpty()) {
// if a certain queue has been drained, reset the flag
allBuffered = false;
+ logger.trace("Partition {} queue is now empty, allBuffered=false", queue.partition());
} else {
nonEmptyQueuesByTime.offer(queue);
}
// always update the stream-time to the record's timestamp yet to be processed if it is larger
if (record.timestamp > streamTime) {
+ final long oldStreamTime = streamTime;
streamTime = record.timestamp;
recordLatenessSensor.record(0, wallClockTime);
+ logger.trace("Partition {} stream time updated from {} to {}", queue.partition(), oldStreamTime, streamTime);
} else {
- recordLatenessSensor.record(streamTime - record.timestamp, wallClockTime);
+ final long lateness = streamTime - record.timestamp;
+ recordLatenessSensor.record(lateness, wallClockTime);
}
}
+ } else {
+ logger.trace("Partition pulling nextRecord: no queue available, totalBuffered={}", totalBuffered);
Review Comment:
Should `totalBuffered` be zero each time we hit this log line?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -450,7 +450,7 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean)
}
hasPendingTxCommit = eosEnabled;
- log.debug("Prepared {} task for committing", state());
+ log.debug("Prepared {} task {} for committing", state(), id);
Review Comment:
Don't we have the task-id already in the log prefix? (Similar elsewhere)
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -733,13 +733,15 @@ public boolean isProcessable(final long wallClockTime) {
if (hasPendingTxCommit) {
// if the task has a pending TX commit, we should just retry the commit but not process any records
// thus, the task is not processable, even if there is available data in the record queue
+ log.debug("Stream task {} has a pending transaction commit, skip processing it.", id());
return false;
}
final boolean readyToProcess = partitionGroup.readyToProcess(wallClockTime);
if (!readyToProcess) {
if (timeCurrentIdlingStarted.isEmpty()) {
timeCurrentIdlingStarted = Optional.of(wallClockTime);
}
+ log.debug("Task {} started idling at time {}", id, timeCurrentIdlingStarted.get());
} else {
timeCurrentIdlingStarted = Optional.empty();
Review Comment:
Should we also add a "resumed" log similar to "started idling" ?
--
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