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_r260980813
##########
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() {
+ @Override
+ public Collection<Entry<?, ?>> process(IncomingMessageEnvelope
message, KeyValueStore store) {
+ return ImmutableList.of(new Entry<>(keySerde.fromBytes((byte[])
message.getKey()), msgSerde.fromBytes((byte[]) message.getMessage())));
+ }
+ });
+ }
+ });
+
+ return sideInputStoresToProcessors;
+ }
+
+ // Create task side input storage managers, one per task, index by the SSP
they are responsible for consuming
+ private Map<SystemStreamPartition, TaskSideInputStorageManager>
createSideInputStorageManagers(Clock clock) {
+
+ Map<SystemStreamPartition, TaskSideInputStorageManager>
sideInputStorageManagers = new HashMap<>();
+
+ containerModel.getTasks().forEach((taskName, taskModel) -> {
+
+ if (!sideInputSystemStreams.isEmpty()) {
Review comment:
can you move it outside the forEach to short circuit earlier?
----------------------------------------------------------------
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
