@Udf
  public List<String> keys(@UdfParameter final String jsonObj) {
    if (jsonObj == null) {
      return null;
    }

    final JsonNode node = UdfJsonMapper.parseJson(jsonObj);
    if (node.isMissingNode() || !node.isObject()) {
      return null;
    }

    final List<String> ret = new ArrayList<>();
    node.fieldNames().forEachRemaining(ret::add);
    return ret;
  }
@Test(expected = KsqlFunctionException.class)
  public void shouldReturnNullForInvalidJson() {
    udf.keys("abc");
  }