@Test
  public void testSendTo() {
    MyMapFunction mapFn = new MyMapFunction();
    StreamApplication app = 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> isd = ksd.getInputDescriptor(PROFILE_STREAM, new NoOpSerde<>());

      appDesc.getInputStream(isd).map(mapFn).sendTo(table);
    };

    InMemorySystemDescriptor isd = new InMemorySystemDescriptor(SYSTEM_NAME);
    InMemoryInputDescriptor<Profile> profileStreamDesc = isd.getInputDescriptor(PROFILE_STREAM, new NoOpSerde<>());

    int numProfilesPerPartition = 10;
    int numInputPartitions = 4;
    Map<Integer, List<Profile>> inputProfiles =
        TestTableData.generatePartitionedProfiles(numProfilesPerPartition * numInputPartitions, numInputPartitions);
    TestRunner.of(app).addInputStream(profileStreamDesc, inputProfiles).run(Duration.ofSeconds(10));

    for (int i = 0; i < numInputPartitions; i++) {
      MyMapFunction mapFnCopy = MyMapFunction.getMapFunctionByTask(String.format("Partition %d", i));
      assertEquals(numProfilesPerPartition, mapFnCopy.received.size());
      mapFnCopy.received.forEach(p -> assertNotNull(mapFnCopy.table.get(p.getMemberId())));
    }
  }
This test class tests sendTo() and join() for local tables
samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableEndToEnd.java