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_r260980336
##########
File path:
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java
##########
@@ -330,14 +486,110 @@ private StorageEngine createStore(String storeName,
TaskName taskName, TaskModel
storeMetricsRegistry, changeLogSystemStreamPartition, jobContext,
containerContext, storeMode);
}
+
+ // Create side input store processors, one per store per task
+ private Map<TaskName, Map<String, SideInputsProcessor>>
createSideInputProcessors(StorageConfig config, ContainerModel containerModel,
+ Map<String, Set<SystemStream>> sideInputSystemStreams, Map<TaskName,
TaskInstanceMetrics> taskInstanceMetrics) {
+
+ Map<TaskName, Map<String, SideInputsProcessor>>
sideInputStoresToProcessors = new HashMap<>();
+ getTasks(containerModel, TaskMode.Active).forEach((taskName, taskModel) ->
{
+ sideInputStoresToProcessors.put(taskName, new HashMap<>());
+ for (String storeName : sideInputSystemStreams.keySet()) {
+ if
(config.getSideInputsProcessorSerializedInstance(storeName).isDefined()) {
+ sideInputStoresToProcessors.get(taskName)
+ .put(storeName, SerdeUtils.deserialize("Side Inputs Processor",
+
config.getSideInputsProcessorSerializedInstance(storeName).get()));
+ } else {
+ sideInputStoresToProcessors.get(taskName)
+ .put(storeName,
Util.getObj(config.getSideInputsProcessorFactory(storeName).get(),
+
SideInputsProcessorFactory.class).getSideInputsProcessor(config,
+ taskInstanceMetrics.get(taskName).registry()));
+ }
+ }
+ });
+
+ // creating identity sideInputProcessor for stores of standbyTasks
+ getTasks(containerModel, TaskMode.Standby).forEach((taskName, taskModel)
-> {
+ sideInputStoresToProcessors.put(taskName, new HashMap<>());
+ for (String storeName : sideInputSystemStreams.keySet()) {
+
+ // have to use the right serde because the sideInput stores are
created
+ Serde keySerde = serdes.get(new
StorageConfig(config).getStorageKeySerde(storeName).get());
+ Serde msgSerde = serdes.get(new
StorageConfig(config).getStorageMsgSerde(storeName).get());
+ sideInputStoresToProcessors.get(taskName).put(storeName, new
SideInputsProcessor() {
Review comment:
nit: extract into constant since its okay to reuse the same identity
processor for all stores.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
