lakshmi-manasa-g commented on issue #1258: SAMZA-2275: Azure Blob System Producer: fixes for high level jobs URL: https://github.com/apache/samza/pull/1258#issuecomment-579534309 > How does `getSystemStreamMetadata` get used? Based on your description, it sounds like `getSystemStreamMetadata` gets called, so does that mean something needs to use the results from it? If you return an empty map, could that be misleading to the logic that uses `getSystemStreamMetadata`? **The exact stack trace of this method being called for the high-level sample job** `at org.apache.samza.system.azureblob.AzureBlobSystemAdmin.getSystemStreamMetadata(AzureBlobSystemAdmin.java:44) at org.apache.samza.execution.StreamManager.getStreamPartitionCounts(StreamManager.java:93) at org.apache.samza.execution.ExecutionPlanner.setInputAndOutputStreamPartitionCount(ExecutionPlanner.java:197) at org.apache.samza.execution.ExecutionPlanner.plan(ExecutionPlanner.java:86) at org.apache.samza.execution.JobPlanner.getExecutionPlan(JobPlanner.java:101) at org.apache.samza.execution.RemoteJobPlanner.prepareJobs(RemoteJobPlanner.java:57) at org.apache.samza.runtime.RemoteApplicationRunner.run(RemoteApplicationRunner.java:73) at org.apache.samza.runtime.ApplicationRunnerUtil.invoke(ApplicationRunnerUtil.java:49) at org.apache.samza.runtime.ApplicationRunnerMain.main(ApplicationRunnerMain.java:53)` **Usage of `getSystemStreamMetadata` in this stack trace:** In the stack trace above, this method is used to get the number of partitions (=size of the map) of a stream in the job graph (all input and output streams). The partition counts are added to the job graph and then used to validate joins and calculate partitions for intermediate streams. Hence, returning an empty map for AzureBlobSystem will result in the partition count being 0 which is does not lead to any incorrect behavior. **Other usages (directly or transitively) of `getSystemStreamMetadata`:** 1. Direct uses: CoordinatorStreamStore, CoordinatorStreamSystemConsumer, StreamManager.getPartitionCounts, StreamMetadataCache, KafkaSystemAdmin, KafkaCheckpointManager, TestRunner, SytemConsumerBench - of these only the StreamManager is relevant cause Azure Blob can not be as CoordinatorStream (as it has no consumer). StreamManager usage is described above. 2. Transitive uses (via SystemAdmin.getSSPMetadata and SystemAdmin.getSystemStreamPartitionCounts) : NonTransactionalStateTaskStorageManager.getNewestChangelogSSPOffsets, TransactionalStateTaskStorageManager.getNewestChangelogSSPOffsets and SSPMetadataCache. None of these are relevant for Azure Blob. **Azure blob Storage as a system in samza:** 1. There is no concept of a partition in Azure Blob Storage - it is mimicked through virtual folders (basically “/“) in the blob name. 2. Samza's SystemStreamMetadata consists of StreamName and 3 offsets (oldest, newest and upcoming) which are not relevant for an Azure blob. For all the above stated reasons, I believe it is acceptable to return an empty map for this method in AzureBlobSystemAdmin.
---------------------------------------------------------------- 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] With regards, Apache Git Services
