Hello vbm,
I am working on the exact same problem. Did you find the solution for the
same.
I am using following code in my client application which will listen to
kafka connect (confluent).
I have one to one mapping for kafka topic and ignite cache. When there is an
insert into db, the kafka listener listens that and using gson library i am
converting json to object and the stmr.addData() works fine. But while
updating the value in db, i am facing marshller error.I tried to use
cache.put() method ,but it gives me cachewriteexception .
@KafkaListener(topics = { "kafka-Users" })
public void listenUsers(String message) {
logger.error(message);
ObjectMapper mapper = new ObjectMapper();
JsonNode rootNode;
try {
rootNode = mapper.readTree(message);
Users user = new Users();
IgniteDataStreamer<Long, Users> stmr =
ignite.dataStreamer(IgniteProperties.USERS_CACHE.getName());
// stmr.allowOverwrite(true);
/*
* stmr.receiver(new StreamTransformer<Long, Users>() {
*
* @Override public Object process(MutableEntry<Long,
Users> entry,
Object...
* arguments) throws EntryProcessorException { return
null; }
*
* });
*/
/*
* stmr.receiver(StreamTransformer.from((e, arg) -> {
Users val =
e.getValue();
* System.out.println(val+" user from reciever
$$$$$$$$$"); return null;
}));
*/
Gson gson = new
GsonBuilder().setFieldNamingPolicy(FieldNamingPolicy.UPPER_CAMEL_CASE).create();
user =
gson.fromJson(rootNode.get("payload").toString(), Users.class);
stmr.addData(rootNode.get("payload").get("UsersKey").asLong(), user);
stmr.flush(); //
// stmr.allowOverwrite(true);
} catch (Exception e) {
e.printStackTrace();
}
}
can you please share your solution for the same.
Thanks,
Om Thacker
--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/
