@Override
            public Collection<Entry<?, ?>> process(IncomingMessageEnvelope message, KeyValueStore store) {
              // Ignore message if the key is null
              if (message.getKey() == null) {
                return ImmutableList.of();
              } else {
                // Skip serde if the message is null
                return ImmutableList.of(new Entry<>(keySerde.fromBytes((byte[]) message.getKey()),
                    message.getMessage() == null ? null : msgSerde.fromBytes((byte[]) message.getMessage())));
              }
            }
Add all sideInputs to a map of maps, indexed first by taskName, then by sideInput store name.

@param sideInputSystemStreams the map of store to sideInput system stream
@param changelogSystemStreams the map of store to changelog system stream
@param containerModel the containerModel to use
@return taskSideInputSSPs map
samza-core/src/main/scala/org/apache/samza/storage/SideInputsManager.java