@Override
    public void process(IncomingMessageEnvelope message, MessageCollector collector, TaskCoordinator coordinator) {
      if (message.getMessage() instanceof Profile) {
        Profile profile = (Profile) message.getMessage();
        profileViewTable.put(profile.getMemberId(), profile);
      } else if (message.getMessage() instanceof PageView) {
        PageView pageView = (PageView) message.getMessage();
        Profile profile = profileViewTable.get(pageView.getMemberId());
        if (profile != null) {
          System.out.println("Joining Page View ArticleView by " + profile.getMemberId());
          collector.send(new OutgoingMessageEnvelope(new SystemStream("test", "EnrichedPageView"), null, null,
              new TestTableData.EnrichedPageView(pageView.getPageKey(), pageView.getMemberId(), profile.getCompany())));
        }
      }
    }
Samza job logic expects integers, but doubles are passed here which results in failure
samza-test/src/test/java/org/apache/samza/test/framework/StreamTaskIntegrationTest.java