@Override
    public void describe(StreamApplicationDescriptor appDesc) {
      KVSerde<Integer, Profile> profileKVSerde = KVSerde.of(new IntegerSerde(), new ProfileJsonSerde());
      KVSerde<Integer, PageView> pageViewKVSerde = KVSerde.of(new IntegerSerde(), new PageViewJsonSerde());

      PageViewToProfileJoinFunction joinFn1 = new PageViewToProfileJoinFunction();
      PageViewToProfileJoinFunction joinFn2 = new PageViewToProfileJoinFunction();

      Table<KV<Integer, Profile>> profileTable = appDesc.getTable(new InMemoryTableDescriptor<>("t1", profileKVSerde));

      DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor(SYSTEM_NAME);
      GenericInputDescriptor<Profile> profileISD1 = ksd.getInputDescriptor(PROFILE_STREAM + "1", new NoOpSerde<>());
      profileISD1.shouldBootstrap();
      GenericInputDescriptor<Profile> profileISD2 = ksd.getInputDescriptor(PROFILE_STREAM + "2", new NoOpSerde<>());
      profileISD2.shouldBootstrap();
      MessageStream<Profile> profileStream1 = appDesc.getInputStream(profileISD1);
      MessageStream<Profile> profileStream2 = appDesc.getInputStream(profileISD2);

      profileStream1
          .map(m -> {
            sentToProfileTable1.add(m);
            return new KV<>(m.getMemberId(), m);
          })
          .sendTo(profileTable);
      profileStream2
          .map(m -> {
            sentToProfileTable2.add(m);
            return new KV<>(m.getMemberId(), m);
          })
          .sendTo(profileTable);

      GenericInputDescriptor<PageView> pageViewISD1 = ksd.getInputDescriptor(PAGEVIEW_STREAM + "1", new NoOpSerde<>());
      GenericInputDescriptor<PageView> pageViewISD2 = ksd.getInputDescriptor(PAGEVIEW_STREAM + "2", new NoOpSerde<>());
      MessageStream<PageView> pageViewStream1 = appDesc.getInputStream(pageViewISD1);
      MessageStream<PageView> pageViewStream2 = appDesc.getInputStream(pageViewISD2);

      pageViewStream1
          .partitionBy(PageView::getMemberId, v -> v, pageViewKVSerde, "p1")
          .join(profileTable, joinFn1)
          .sink((m, collector, coordinator) -> joinedPageViews1.add(m));

      pageViewStream2
          .partitionBy(PageView::getMemberId, v -> v, pageViewKVSerde, "p2")
          .join(profileTable, joinFn2)
          .sink((m, collector, coordinator) -> joinedPageViews2.add(m));
    }
This test class tests sendTo() and join() for local tables
samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableEndToEnd.java