@Override
  public void describe(StreamApplicationDescriptor appDescriptor) {
    KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");

    KafkaInputDescriptor<OrderRecord> orderStreamDescriptor =
        trackingSystem.getInputDescriptor("orders", new JsonSerdeV2<>(OrderRecord.class));
    KafkaInputDescriptor<ShipmentRecord> shipmentStreamDescriptor =
        trackingSystem.getInputDescriptor("shipments", new JsonSerdeV2<>(ShipmentRecord.class));
    KafkaOutputDescriptor<KV<String, FulfilledOrderRecord>> fulfilledOrdersStreamDescriptor =
        trackingSystem.getOutputDescriptor("fulfilledOrders",
            KVSerde.of(new StringSerde(), new JsonSerdeV2<>(FulfilledOrderRecord.class)));

    appDescriptor.getInputStream(orderStreamDescriptor)
        .join(appDescriptor.getInputStream(shipmentStreamDescriptor), new MyJoinFunction(),
            new StringSerde(), new JsonSerdeV2<>(OrderRecord.class), new JsonSerdeV2<>(ShipmentRecord.class),
            Duration.ofMinutes(1), "join")
        .map(fulFilledOrder -> KV.of(fulFilledOrder.orderId, fulFilledOrder))
        .sendTo(appDescriptor.getOutputStream(fulfilledOrdersStreamDescriptor));

  }
Simple 2-way stream-to-stream join example
samza-test/src/main/java/org/apache/samza/example/OrderShipmentJoinExample.java