mynameborat commented on a change in pull request #912: SEP-19 : Refactoring
sideInputs from SamzaContainer to ContainerStorageManager
URL: https://github.com/apache/samza/pull/912#discussion_r260979620
##########
File path:
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java
##########
@@ -222,55 +377,53 @@ public ContainerStorageManager(ContainerModel
containerModel, StreamMetadataCach
for (String storeName : storageEngineFactories.keySet()) {
+ StorageEngineFactory.StoreMode storeMode =
this.sideInputSystemStreams.containsKey(storeName) ?
+ StorageEngineFactory.StoreMode.ReadWrite :
StorageEngineFactory.StoreMode.BulkLoad;
+
StorageEngine storageEngine =
- createStore(storeName, taskName, taskModel, jobContext,
containerContext, storageEngineFactories,
- changelogSystemStreams, serdes, taskInstanceMetrics,
taskInstanceCollectors, storeMode);
+ createStore(storeName, taskName, taskModel, jobContext,
containerContext, storageEngineFactories, serdes, taskInstanceMetrics,
taskInstanceCollectors, storeMode);
// add created store to map
taskStores.get(taskName).put(storeName, storageEngine);
- LOG.info("Created store {} for task {}", storeName, taskName);
+ LOG.info("Created store {} for task {} in mode {}", storeName,
taskName, storeMode);
}
}
return taskStores;
}
/**
- * Recreate all persistent stores in ReadWrite mode.
+ * Recreate all non-sideInput persistent stores in ReadWrite mode.
*
*/
private void recreatePersistentTaskStoresInReadWriteMode(ContainerModel
containerModel, JobContext jobContext,
ContainerContext containerContext, Map<String,
StorageEngineFactory<Object, Object>> storageEngineFactories,
- Map<String, SystemStream> changelogSystemStreams, Map<String,
Serde<Object>> serdes,
- Map<TaskName, TaskInstanceMetrics> taskInstanceMetrics,
+ Map<String, Serde<Object>> serdes, Map<TaskName, TaskInstanceMetrics>
taskInstanceMetrics,
Map<TaskName, TaskInstanceCollector> taskInstanceCollectors) {
// iterate over each task and each storeName
for (Map.Entry<TaskName, TaskModel> task :
containerModel.getTasks().entrySet()) {
TaskName taskName = task.getKey();
TaskModel taskModel = task.getValue();
+ Map<String, StorageEngine> nonSideInputStores =
getNonSideInputStores(taskName);
- for (String storeName : storageEngineFactories.keySet()) {
+ for (String storeName : nonSideInputStores.keySet()) {
- // if this store has been already created in the taskStores, then
re-create and overwrite it only if it is a persistentStore
- if (this.taskStores.get(taskName).containsKey(storeName) &&
this.taskStores.get(taskName)
- .get(storeName)
- .getStoreProperties()
- .isPersistedToDisk()) {
+ // if this store has been already then re-create and overwrite it only
if it is a
Review comment:
nit: missing "created in the taskStores"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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
With regards,
Apache Git Services
