You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@ignite.apache.org by Om Thacker <om...@gmail.com> on 2019/06/01 10:54:54 UTC
Re: Questions on IgniteDataStreamer
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/
Re: Questions on IgniteDataStreamer
Posted by Ilya Kasnacheev <il...@gmail.com>.
Hello!
Can you please at least share the exceptions you are getting?
Regards,
--
Ilya Kasnacheev
сб, 1 июн. 2019 г. в 13:55, Om Thacker <om...@gmail.com>:
> 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/
>