rmatharu 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_r261025331
##########
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:
Each store is created with its respective configured serdes, so this
processor needs to use that serde.
----------------------------------------------------------------
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
