SAMZA-1516: Another round of issues found by BEAM tests

[xinyuiscool]

A couple of more fixes: 1. fix a bug of identifying input streams for an operator. 2. for partitionBy, set the partitionKey to 0L when key is null.

[fredji97]

after checking with Xinyu, Kafka has a tricky behavior to do round robin when both the partition key and key are null. It might be better to add a line of comment in code as well for future maintenance reference.

[vjagadish1989]

This behavior in Kafka is pretty reasonable. A null key actually means that a key is absent (as opposed to a key being null) and the message is non-keyed. Imagine all non-keyed messages going to the same partition 0? That would be very sub-optimal. For stream-processing use-cases, we want to use null for grouping by key. Hence, we need this to get around this.

[vjagadish1989]

Are we doing a dual registration here? First in the if clause, and then in the else clause? Can we consider consolidating the recursion logic then?

[xinyuiscool]

Sure, I can extract the registerInputStream outside the if .. else. SHould be straightforward to do.

[vjagadish1989]

wondering if this logic is needed here? Is the following equivalent?

I'd flip the statement that adds it to the map, and add it to the operatorImpls after the recursive call returns?

  OperatorImpl createAndRegisterOperatorImpl(OperatorSpec prevOperatorSpec, OperatorSpec operatorSpec,
      SystemStream inputStream, Config config, TaskContext context) {
    if (!operatorImpls.containsKey(operatorSpec.getOpId()) || operatorSpec instanceof JoinOperatorSpec) {
      OperatorImpl operatorImpl = createOperatorImpl(prevOperatorSpec, operatorSpec, config, context);
      operatorImpl.init(config, context);
      operatorImpl.registerInputStream(inputStream);

      Collection<OperatorSpec> registeredSpecs = operatorSpec.getRegisteredOperatorSpecs();
      registeredSpecs.forEach(registeredSpec -> {
          OperatorImpl nextImpl = createAndRegisterOperatorImpl(operatorSpec, registeredSpec, inputStream, config, context);
          operatorImpl.registerNextOperator(nextImpl);
        });
      operatorImpls.put(operatorImpl.getOpImplId(), operatorImpl);
      return operatorImpl;
    } else {
      return operatorImpls.get(operatorSpec.getOpId());
    }
  }

[xinyuiscool]

This logic is needed since I need to further traverse the graph even if the operator is already registered. That way the input stream can be registered to the downstream operators.