@SuppressWarnings({ "unchecked", "rawtypes" })
  public void testSendMessage() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {

    //check a correct message
    assertEquals(0, systemProducer.getEnvelopes().size());
    coordinatorStreamWriter.sendMessage("set-config", "key0", "value0");
    assertEquals(1, systemProducer.getEnvelopes().size());

    //check invalid input is handled
    boolean exceptionHappened = false;
    try {
      coordinatorStreamWriter.sendMessage("invalid-type", "key-invalid", "value-invalid");
    } catch (IllegalArgumentException e) {
      exceptionHappened = true;
    }
    assertTrue(exceptionHappened);
    assertEquals(1, systemProducer.getEnvelopes().size());


    //check sendSetConfigMessage method works correctly
    Class[] sendArgs = {String.class, String.class};
    Method sendSetConfigMethod = coordinatorStreamWriter.getClass().getDeclaredMethod("sendSetConfigMessage", sendArgs);
    sendSetConfigMethod.setAccessible(true);
    sendSetConfigMethod.invoke(coordinatorStreamWriter, "key1", "value1");
    assertEquals(2, systemProducer.getEnvelopes().size());


    //check the messages are correct
    List<OutgoingMessageEnvelope> envelopes = systemProducer.getEnvelopes();
    OutgoingMessageEnvelope envelope0 = envelopes.get(0);
    OutgoingMessageEnvelope envelope1 = envelopes.get(1);
    TypeReference<Object[]> keyRef = new TypeReference<Object[]>() {
    };
    TypeReference<Map<String, Object>> msgRef = new TypeReference<Map<String, Object>>() {
    };
    assertEquals(2, envelopes.size());

    assertEquals("key0", deserialize((byte[]) envelope0.getKey(), keyRef)[CoordinatorStreamMessage.KEY_INDEX]);
    Map<String, String> values = (Map<String, String>) deserialize((byte[]) envelope0.getMessage(), msgRef).get("values");
    assertEquals("value0", values.get("value"));

    assertEquals("key1", deserialize((byte[]) envelope1.getKey(), keyRef)[CoordinatorStreamMessage.KEY_INDEX]);
    values = (Map<String, String>) deserialize((byte[]) envelope1.getMessage(), msgRef).get("values");
    assertEquals("value1", values.get("value"));
  }
This class is a unit test for the CoordinatorStreamWriter class.
samza-core/src/test/java/org/apache/samza/coordinator/stream/TestCoordinatorStreamWriter.java