private PartialJoinFunction<Object, Object, Object, Object> createLeftJoinFn(JoinOperatorSpec joinOpSpec) {
    return new PartialJoinFunction<Object, Object, Object, Object>() {
      private final JoinFunction joinFn = joinOpSpec.getJoinFn();
      private KeyValueStore<Object, TimestampedValue<Object>> leftStreamState;

      @Override
      public Object apply(Object m, Object om) {
        return joinFn.apply(m, om);
      }

      @Override
      public Object getKey(Object message) {
        return joinFn.getFirstKey(message);
      }

      @Override
      public KeyValueStore<Object, TimestampedValue<Object>> getState() {
        return leftStreamState;
      }

      @Override
      public void init(Context context) {
        String leftStoreName = joinOpSpec.getLeftOpId();
        leftStreamState =
            (KeyValueStore<Object, TimestampedValue<Object>>) context.getTaskContext().getStore(leftStoreName);

        // user-defined joinFn should only be initialized once, so we do it only in left partial join function.
        joinFn.init(context);
      }

      @Override
      public void close() {
        // joinFn#close() must only be called once, so we do it it only in left partial join function.
        joinFn.close();
      }
    };
  }
Creates a new {@link OperatorImpl} instance for the provided {@link OperatorSpec}.

@param prevOperatorSpec the original {@link OperatorSpec} that produces output for {@code operatorSpec} from {@link OperatorSpecGraph}
@param operatorSpec  the original {@link OperatorSpec} from {@link OperatorSpecGraph}
@param context  the {@link Context} required to instantiate operators
@return  the {@link OperatorImpl} implementation instance
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImplGraph.java