@Test
  public void testSyncTaskWithMultiplePartitionMultithreadedWithCustomIME() throws Exception {
    Map<Integer, List<KV>> inputPartitionData = new HashMap<>();
    Map<Integer, List<KV>> inputPartitionIME = new HashMap<>();
    Map<Integer, List<Integer>> expectedOutputPartitionData = new HashMap<>();
    genData(inputPartitionData, expectedOutputPartitionData);

    for (Map.Entry<Integer, List<KV>> entry: inputPartitionData.entrySet()) {
      Integer partitionId = entry.getKey();
      List<KV> messages = entry.getValue();
      SystemStreamPartition ssp = new SystemStreamPartition("test", "input", new Partition(partitionId));
      inputPartitionIME.put(partitionId, new ArrayList<>());
      int offset = 0;
      for (KV message: messages) {
        IncomingMessageEnvelope ime = new IncomingMessageEnvelope(ssp, String.valueOf(offset++), message.key, message.getValue());
        inputPartitionIME.get(partitionId).add(KV.of(message.key, ime));
      }
    }
    syncTaskWithMultiplePartitionMultithreadedHelper(inputPartitionIME, expectedOutputPartitionData);
  }
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