@Override
    public void describe(StreamApplicationDescriptor appDesc) {
      Table<KV<Integer, Profile>> table = appDesc.getTable(
          new InMemoryTableDescriptor<>("t1", KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())));
      DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor(SYSTEM_NAME);
      GenericInputDescriptor<Profile> profileISD = ksd.getInputDescriptor(PROFILE_STREAM, new NoOpSerde<>());
      profileISD.shouldBootstrap();
      appDesc.getInputStream(profileISD)
          .map(m -> new KV<>(m.getMemberId(), m))
          .sendTo(table);

      GenericInputDescriptor<PageView> pageViewISD = ksd.getInputDescriptor(PAGEVIEW_STREAM, new NoOpSerde<>());
      appDesc.getInputStream(pageViewISD)
          .map(pv -> {
            received.add(pv);
            return pv;
          })
          .partitionBy(PageView::getMemberId, v -> v, KVSerde.of(new IntegerSerde(), new PageViewJsonSerde()), "p1")
          .join(table, new PageViewToProfileJoinFunction())
          .sink((m, collector, coordinator) -> joined.add(m));
    }
This test class tests sendTo() and join() for local tables
samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableEndToEnd.java