@Test
  public void testDualStreamTableJoin() {
    int totalPageViews = 40;
    int partitionCount = 4;
    Map<Integer, List<PageView>> inputPageViews1 =
        TestTableData.generatePartitionedPageViews(totalPageViews, partitionCount);
    Map<Integer, List<PageView>> inputPageViews2 =
        TestTableData.generatePartitionedPageViews(totalPageViews, partitionCount);
    // 10 is the max member id for page views
    int numProfiles = 10;
    Map<Integer, List<Profile>> inputProfiles1 = TestTableData.generatePartitionedProfiles(numProfiles, partitionCount);
    Map<Integer, List<Profile>> inputProfiles2 = TestTableData.generatePartitionedProfiles(numProfiles, partitionCount);
    InMemorySystemDescriptor isd = new InMemorySystemDescriptor(SYSTEM_NAME);
    InMemoryInputDescriptor<PageView> pageViewStreamDesc1 = isd
        .getInputDescriptor(PAGEVIEW_STREAM + "1", new NoOpSerde<>());
    InMemoryInputDescriptor<PageView> pageViewStreamDesc2 = isd
        .getInputDescriptor(PAGEVIEW_STREAM + "2", new NoOpSerde<>());
    InMemoryInputDescriptor<Profile> profileStreamDesc1 = isd
        .getInputDescriptor(PROFILE_STREAM + "1", new NoOpSerde<>());
    InMemoryInputDescriptor<Profile> profileStreamDesc2 = isd
        .getInputDescriptor(PROFILE_STREAM + "2", new NoOpSerde<>());

    TestRunner.of(new DualStreamTableJoinApp())
        .addInputStream(pageViewStreamDesc1, inputPageViews1)
        .addInputStream(pageViewStreamDesc2, inputPageViews2)
        .addInputStream(profileStreamDesc1, inputProfiles1)
        .addInputStream(profileStreamDesc2, inputProfiles2)
        .run(Duration.ofSeconds(10));

    assertEquals(numProfiles, DualStreamTableJoinApp.sentToProfileTable1.size());
    assertEquals(numProfiles, DualStreamTableJoinApp.sentToProfileTable2.size());
    assertEquals(totalPageViews, DualStreamTableJoinApp.joinedPageViews1.size());
    assertEquals(totalPageViews, DualStreamTableJoinApp.joinedPageViews2.size());
    assertNotNull(DualStreamTableJoinApp.joinedPageViews1.get(0));
    assertNotNull(DualStreamTableJoinApp.joinedPageViews2.get(0));
  }
This test class tests sendTo() and join() for local tables
samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableEndToEnd.java