spena commented on a change in pull request #10211:
URL: https://github.com/apache/kafka/pull/10211#discussion_r586578441
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
##########
@@ -282,6 +284,21 @@ public boolean commitNeeded() {
return Collections.unmodifiableMap(stateMgr.changelogOffsets());
}
+ @Override
+ public Map<TopicPartition, Long> getCommittedOffsets() {
+ return new HashMap<>();
+ }
+
+ @Override
+ public Map<TopicPartition, Long> getHighWaterMark() {
+ return new HashMap<>();
+ }
Review comment:
Should this two return a `Collections.emptyMap()` instead? Or is it
expected that the returned map is modifiable?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/TaskMetadata.java
##########
@@ -32,10 +34,22 @@
private final Set<TopicPartition> topicPartitions;
+ private final Map<TopicPartition, Long> committedOffsets;
+
+ private final Map<TopicPartition, Long> endOffsets;
+
+ private Optional<Long> timeCurrentIdlingStarted;
+
public TaskMetadata(final String taskId,
- final Set<TopicPartition> topicPartitions) {
+ final Set<TopicPartition> topicPartitions,
+ final Map<TopicPartition, Long> committedOffsets,
+ final Map<TopicPartition, Long> endOffsets,
+ final Optional<Long> timeCurrentIdlingStarted) {
this.taskId = taskId;
this.topicPartitions = Collections.unmodifiableSet(topicPartitions);
+ this.committedOffsets = committedOffsets;
+ this.endOffsets = endOffsets;
Review comment:
Should these be wrapped with `Collections.unmodifiableMap()` or
`ImmutableMap.copyOf()`? If the two maps are not expected to be modified after
passing them as parameter, then I suggest `ImmutableMap.copyOf()`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
