@Override
    public void describe(StreamApplicationDescriptor appDescriptor) {
      Table<KV<Integer, TestTableData.Profile>> table = appDescriptor.getTable(
          new RocksDbTableDescriptor<Integer, TestTableData.Profile>("profile-view-store",
              KVSerde.of(new IntegerSerde(), new TestTableData.ProfileJsonSerde())));

      KafkaSystemDescriptor ksd = new KafkaSystemDescriptor("test");

      KafkaInputDescriptor<KV<String, TestTableData.Profile>> profileISD =
          ksd.getInputDescriptor("Profile", KVSerde.of(new StringSerde(), new JsonSerdeV2<>()));

      KafkaInputDescriptor<KV<String, TestTableData.PageView>> pageViewISD =
          ksd.getInputDescriptor("PageView", KVSerde.of(new StringSerde(), new JsonSerdeV2<>()));
      KafkaOutputDescriptor<TestTableData.EnrichedPageView> enrichedPageViewOSD =
          ksd.getOutputDescriptor("EnrichedPageView", new JsonSerdeV2<>());

      appDescriptor.getInputStream(profileISD)
          .map(m -> new KV(m.getValue().getMemberId(), m.getValue()))
          .sendTo(table)
          .sink((kv, collector, coordinator) -> {
            LOG.info("Inserted Profile with Key: {} in profile-view-store", kv.getKey());
          });

      OutputStream<TestTableData.EnrichedPageView> outputStream = appDescriptor.getOutputStream(enrichedPageViewOSD);
      appDescriptor.getInputStream(pageViewISD)
          .partitionBy(pv -> pv.getValue().getMemberId(),  pv -> pv.getValue(), KVSerde.of(new IntegerSerde(), new JsonSerdeV2<>(TestTableData.PageView.class)), "p1")
          .join(table, new PageViewToProfileJoinFunction())
          .sendTo(outputStream)
          .map(TestTableData.EnrichedPageView::getPageKey)
          .sink((joinPageKey, collector, coordinator) -> {
            collector.send(new OutgoingMessageEnvelope(new SystemStream("test", "JoinPageKeys"), null, null, joinPageKey));
          });

    }
Null page key is passed in input data which should fail filter logic
samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTest.java