@Test
  public void testStreamTableJoin() {
    int totalPageViews = 40;
    int partitionCount = 4;
    Map<Integer, List<PageView>> inputPageViews =
        TestTableData.generatePartitionedPageViews(totalPageViews, partitionCount);
    // 10 is the max member id for page views
    Map<Integer, List<Profile>> inputProfiles =
        TestTableData.generatePartitionedProfiles(10, partitionCount);
    InMemorySystemDescriptor isd = new InMemorySystemDescriptor(SYSTEM_NAME);
    InMemoryInputDescriptor<PageView> pageViewStreamDesc = isd
        .getInputDescriptor(PAGEVIEW_STREAM, new NoOpSerde<>());
    InMemoryInputDescriptor<Profile> profileStreamDesc = isd
        .getInputDescriptor(PROFILE_STREAM, new NoOpSerde<>());

    TestRunner.of(new StreamTableJoinApp())
        .addInputStream(pageViewStreamDesc, inputPageViews)
        .addInputStream(profileStreamDesc, inputProfiles)
        .run(Duration.ofSeconds(10));

    assertEquals(totalPageViews, StreamTableJoinApp.received.size());
    assertEquals(totalPageViews, StreamTableJoinApp.joined.size());
    assertNotNull(StreamTableJoinApp.joined.get(0));
  }
This test class tests sendTo() and join() for local tables
samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableEndToEnd.java