MessageStreamImpl implements MessageStream<M> { @Override public <K, OM, JM> MessageStream<JM> join(MessageStream<OM> otherStream, JoinFunction<? extends K, ? super M, ? super OM, ? extends JM> joinFn, Serde<K> keySerde, Serde<M> messageSerde, Serde<OM> otherMessageSerde, Duration ttl, String userDefinedId) { if (otherStream.equals(this)) throw new SamzaException("Cannot join a MessageStream with itself."); String opId = this.streamAppDesc.getNextOpId(OpCode.JOIN, userDefinedId); OperatorSpec<?, OM> otherOpSpec = ((MessageStreamImpl<OM>) otherStream).getOperatorSpec(); JoinOperatorSpec<K, M, OM, JM> op = OperatorSpecs.createJoinOperatorSpec(this.operatorSpec, otherOpSpec, (JoinFunction<K, M, OM, JM>) joinFn, keySerde, messageSerde, otherMessageSerde, ttl.toMillis(), opId); this.operatorSpec.registerNextOperatorSpec(op); otherOpSpec.registerNextOperatorSpec((OperatorSpec<OM, ?>) op); return new MessageStreamImpl<>(this.streamAppDesc, op); } MessageStreamImpl(StreamApplicationDescriptorImpl streamAppDesc, OperatorSpec<?, M> operatorSpec); @Override MessageStream<TM> map(MapFunction<? super M, ? extends TM> mapFn); @Override MessageStream<M> filter(FilterFunction<? super M> filterFn); @Override MessageStream<TM> flatMap(FlatMapFunction<? super M, ? extends TM> flatMapFn); @Override MessageStream<OM> flatMapAsync(AsyncFlatMapFunction<? super M, ? extends OM> flatMapFn); @Override void sink(SinkFunction<? super M> sinkFn); @Override MessageStream<M> sendTo(OutputStream<M> outputStream); @Override MessageStream<WindowPane<K, WV>> window(Window<M, K, WV> window, String userDefinedId); @Override MessageStream<JM> join(MessageStream<OM> otherStream,
      JoinFunction<? extends K, ? super M, ? super OM, ? extends JM> joinFn,
      Serde<K> keySerde, Serde<M> messageSerde, Serde<OM> otherMessageSerde,
      Duration ttl, String userDefinedId); @Override MessageStream<JM> join(Table<R> table,
      StreamTableJoinFunction<? extends K, ? super M, ? super R, ? extends JM> joinFn, Object ... args); @Override MessageStream<M> merge(Collection<? extends MessageStream<? extends M>> otherStreams); @Override MessageStream<KV<K, V>> partitionBy(MapFunction<? super M, ? extends K> keyExtractor,
      MapFunction<? super M, ? extends V> valueExtractor, KVSerde<K, V> serde, String userDefinedId); @Override MessageStream<KV<K, V>> sendTo(Table<KV<K, V>> table, Object ... args); @Override MessageStream<M> broadcast(Serde<M> serde, String userDefinedId); @VisibleForTesting OperatorSpec<?, M> getOperatorSpec();  }
@Test public void testJoin() { StreamApplicationDescriptorImpl mockGraph = mock(StreamApplicationDescriptorImpl.class); OperatorSpec leftInputOpSpec = mock(OperatorSpec.class); MessageStreamImpl<TestMessageEnvelope> source1 = new MessageStreamImpl<>(mockGraph, leftInputOpSpec); OperatorSpec rightInputOpSpec = mock(OperatorSpec.class); MessageStreamImpl<TestMessageEnvelope> source2 = new MessageStreamImpl<>(mockGraph, rightInputOpSpec); JoinFunction<String, TestMessageEnvelope, TestMessageEnvelope, TestOutputMessageEnvelope> mockJoinFn = mock(JoinFunction.class); Duration joinTtl = Duration.ofMinutes(1); source1.join(source2, mockJoinFn, mock(Serde.class), mock(Serde.class), mock(Serde.class), joinTtl, "j1"); ArgumentCaptor<OperatorSpec> leftRegisteredOpCaptor = ArgumentCaptor.forClass(OperatorSpec.class); verify(leftInputOpSpec).registerNextOperatorSpec(leftRegisteredOpCaptor.capture()); OperatorSpec<?, TestMessageEnvelope> leftRegisteredOpSpec = leftRegisteredOpCaptor.getValue(); ArgumentCaptor<OperatorSpec> rightRegisteredOpCaptor = ArgumentCaptor.forClass(OperatorSpec.class); verify(rightInputOpSpec).registerNextOperatorSpec(rightRegisteredOpCaptor.capture()); OperatorSpec<?, TestMessageEnvelope> rightRegisteredOpSpec = rightRegisteredOpCaptor.getValue(); assertEquals(leftRegisteredOpSpec, rightRegisteredOpSpec); assertEquals(OpCode.JOIN, leftRegisteredOpSpec.getOpCode()); assertTrue(leftRegisteredOpSpec instanceof JoinOperatorSpec); assertEquals(mockJoinFn, ((JoinOperatorSpec) leftRegisteredOpSpec).getJoinFn()); assertEquals(joinTtl.toMillis(), ((JoinOperatorSpec) leftRegisteredOpSpec).getTtlMs()); assertEquals(leftInputOpSpec, ((JoinOperatorSpec) leftRegisteredOpSpec).getLeftInputOpSpec()); assertEquals(rightInputOpSpec, ((JoinOperatorSpec) leftRegisteredOpSpec).getRightInputOpSpec()); }