public synchronized void process(IncomingMessageEnvelope envelope) {
    SystemStreamPartition envelopeSSP = envelope.getSystemStreamPartition();
    String envelopeOffset = envelope.getOffset();

    for (String store: this.sspToStores.get(envelopeSSP)) {
      SideInputsProcessor storeProcessor = this.storeToProcessor.get(store);
      KeyValueStore keyValueStore = (KeyValueStore) this.taskSideInputStorageManager.getStore(store);
      Collection<Entry<?, ?>> entriesToBeWritten = storeProcessor.process(envelope, keyValueStore);

      // TODO: SAMZA-2255: optimize writes to side input stores
      for (Entry entry : entriesToBeWritten) {
        // If the key is null we ignore, if the value is null, we issue a delete, else we issue a put
        if (entry.getKey() != null) {
          if (entry.getValue() != null) {
            keyValueStore.put(entry.getKey(), entry.getValue());
          } else {
            keyValueStore.delete(entry.getKey());
          }
        }
      }
    }

    this.lastProcessedOffsets.put(envelopeSSP, envelopeOffset);
    checkCaughtUp(envelopeSSP, envelopeOffset, SystemStreamMetadata.OffsetType.NEWEST);
  }
Processes the incoming side input message envelope and updates the last processed offset for its SSP.
Synchronized inorder to be exclusive with flush().

@param envelope incoming envelope to be processed
samza-core/src/main/java/org/apache/samza/storage/TaskSideInputHandler.java