@Override
  public T fromBytes(byte[] bytes) {
    if (bytes != null) {
      String str;
      try {
        str = new String(bytes, "UTF-8");
      } catch (UnsupportedEncodingException e) {
        throw new SamzaException("Error deserializing data", e);
      }

      try {
        if (clazz != null) {
          return mapper.readValue(str, clazz);
        } else {
          return mapper.readValue(str, new TypeReference<T>() { });
        }
      } catch (Exception e) {
        LOG.debug("Error deserializing data: " + str, e);
        throw new SamzaException("Error deserializing data", e);
      }
    } else {
      return null;
    }
  }
Constructs a JsonSerdeV2 that (de)serializes POJOs of class {@code clazz}.

@param clazz the class of the POJO being (de)serialized.
samza-api/src/main/java/org/apache/samza/serializers/JsonSerdeV2.java