Here is the code. I can only get log in the constructor and configure method.
import org.json.simple.JSONObject;import
org.apache.kafka.common.errors.SerializationException;import
org.apache.kafka.common.serialization.Serializer;
import java.util.Map;import org.slf4j.Logger;import org.slf4j.LoggerFactory
public class JsonSerializer implements Serializer<JSONObject> {
private static final Logger LOG =
LoggerFactory.getLogger(JsonSerializer.class);
/** * Default constructor needed by Kafka */
public JsonSerializer() {
LOG.info("===> JsonSerializer constructor !!");
}
@Override
public void configure(Map<String, ?> config, boolean isKey) {
LOG.info("===> JsonSerializer configure");
}
@Override
public byte[] serialize(String topic, JSONObject data) {
LOG.info("===> JsonSerializer serialize !!");
if (data == null)
return null;
try {
return data.toString().getBytes("utf-8");
} catch (Exception e) {
LOG.error("===> JsonSerializer serialize EXCEPTION");
throw new SerializationException("Error serializing JSON
message", e);
}
}
@Override
public void close() {
LOG.error("===> JsonSerializer close");
}}
2016-11-14 20:01 GMT+08:00 Andrew Xor <[email protected]>:
> Hi,
>
> Since you can't cast one type to another and you are not getting a Null
> exception in order to be better able to help you could you give us the
> implementation of your serializer?
>
> Cheers,
>
> A.
>
> On Mon, Nov 14, 2016 at 9:17 AM, Zhechao Ma <[email protected]>
> wrote:
>
>> Even when I implement my own json serializer, it still throws the similar
>> exception, but no more details for debug.:
>>
>> org.apache.kafka.common.errors.SerializationException: Can't convert
>> value of class org.apache.storm.shade.org.json.simple.JSONObject to
>> xxxxxx specified in value.serializer
>>
>> I add a LOG in the overriding method byte[] serialize(String topic,
>> JSONObject data), and found no logs in worker.log. That is to say this
>> exception is throwed before method serialize is called.
>>
>> 2016-11-07 16:50 GMT+08:00 Zhechao Ma <[email protected]>:
>>
>>> I'm using KafkaBolt to write data to kafka. Tuple to kafka map is
>>> <String, JSONObject>.
>>> I set both *key.serializer* and *value.serializer* as
>>> "org.apache.kafka.common.serialization.StringSerializer". I get the
>>> following Exception:
>>>
>>> org.apache.kafka.common.errors.SerializationException: Can't convert
>>> value of class org.apache.storm.shade.org.json.simple.JSONObject to
>>> class org.apache.kafka.common.serialization.StringSerializer specified
>>> in value.serializer
>>>
>>>
>>> I cannot find other serializers related to JSON, and I'm using storm
>>> 1.0.2 and kafka 0.8.1.1.
>>>
>>> Could anyone help ?
>>>
>>> --
>>> Thanks
>>> Zhechao Ma
>>>
>>
>>
>>
>> --
>> Thanks
>> Zhechao Ma
>>
>
>
--
Thanks
Zhechao Ma
