lakshmi-manasa-g commented on a change in pull request #1385:
URL: https://github.com/apache/samza/pull/1385#discussion_r443848511
##########
File path:
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java
##########
@@ -758,41 +744,50 @@ public void run() {
sideInputSystemConsumers.register(ssp, startingOffset);
taskInstanceMetrics.get(this.sspSideInputHandlers.get(ssp).getTaskName()).addOffsetGauge(
ssp, ScalaJavaUtil.toScalaFunction(() ->
this.sspSideInputHandlers.get(ssp).getLastProcessedOffset(ssp)));
+
sideInputTaskMetrics.get(this.sspSideInputHandlers.get(ssp).getTaskName()).addOffsetGauge(
+ ssp, ScalaJavaUtil.toScalaFunction(() ->
this.sspSideInputHandlers.get(ssp).getLastProcessedOffset(ssp)));
+ }
- SystemStreamMetadata systemStreamMetadata =
streamMetadataCache.getSystemStreamMetadata(ssp.getSystemStream(), false);
- SystemStreamMetadata.SystemStreamPartitionMetadata sspMetadata =
- (systemStreamMetadata == null) ? null :
systemStreamMetadata.getSystemStreamPartitionMetadata().get(ssp.getPartition());
+ Map<TaskName, TaskSideInputHandler> taskSideInputHandlers =
this.sspSideInputHandlers.values().stream()
+ .distinct()
+ .collect(Collectors.toMap(TaskSideInputHandler::getTaskName,
Function.identity()));
- // record a copy of the sspMetadata, to later check if its caught up
- initialSideInputSSPMetadata.put(ssp, sspMetadata);
+ Map<TaskName, RunLoopTask> sideInputTasks = new HashMap<>();
+ this.taskSideInputStoreSSPs.forEach((taskName, storesToSSPs) -> {
+ Set<SystemStreamPartition> taskSSPs =
this.taskSideInputStoreSSPs.get(taskName).values().stream()
+ .flatMap(Set::stream)
+ .collect(Collectors.toSet());
- // check if the ssp is caught to upcoming, even at start
- checkSideInputCaughtUp(ssp, startingOffset,
SystemStreamMetadata.OffsetType.UPCOMING, false);
Review comment:
just confirming: the check is the same now because
StorageManagerUtil.getStartingOffset will get the Upcoming offset and from the
definition of Upcoming and Newest offset - Upcoming > Newest. So, although the
new checkCaughtUp replacing this method is not explicitly using the
OffsetTypes, it is still achieving the same.
##########
File path:
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java
##########
@@ -111,13 +109,10 @@
private static final Logger LOG =
LoggerFactory.getLogger(ContainerStorageManager.class);
private static final String RESTORE_THREAD_NAME = "Samza Restore Thread-%d";
private static final String SIDEINPUTS_READ_THREAD_NAME = "SideInputs Read
Thread";
Review comment:
minor: maybe we can call this thread just "sideInputs thread" as it does
both read and flush?
----------------------------------------------------------------
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:
[email protected]
