kudu-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From José Ribeiro <jose.enes.ribe...@hotmail.com>
Subject Re: AsyncKudu
Date Mon, 09 Apr 2018 17:16:31 GMT
Thank you for the reply. But I got it to write on Kudu.

Still, I don't know why, but I get this error:


"java.lang.AssertionError: Tried to resume the execution of Deferred@2021942318(state=DONE,
result=true, callback=<none>, errback=<none>)) although it's not in state=PAUSED.
 This occurred after the completion of Deferred@1753159143(state=RUNNING, result=Written with
success., callback=<none>, errback=<none>) which was originally returned by callback=com.foo.kudu.KuduDAOImpl$1@3e69008f@1047068815"

The only way I discover to reach kudu was like this:

 public void asyncWrite(String tableName, Foo foo) {
        Deferred<Deferred<Boolean>> res = new Deferred<>();

        res.addCallback(new Callback<Deferred<Deferred<String>>, Deferred<Boolean>>()
{
            @Override
            public Deferred<Deferred<String>> call(Deferred<Boolean> booleanDeferred){
                return booleanDeferred.addCallbacks(new Callback<Deferred<String>,
Boolean>() {
                    @Override
                    public Deferred<String> call(Boolean aBoolean) throws Exception
{
                        if (aBoolean) {
                            Deferred<String> stringDeferred = kuduClient.openTable(tableName)
                                    .addCallbacks(new Callback<String, KuduTable>()
{
                                        @Override
                                        public String call(KuduTable kuduTable) throws KuduException
{
                                            AsyncKuduSession session = kuduClient.newSession();
                                            session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);

                                            Upsert upst = kuduTable.newUpsert();
                                            PartialRow row = upst.getRow();
                                            row.addString("house_id", foo.getId());

                                            session.apply(upst);
                                            session.close();
                                            LOGGER.info("Written with success.");
                                            return "Written with success.";
                                        }
                                    }, (Callback<Object, KuduException>) e -> {
                                        LOGGER.error(e.getMessage());
                                        return e;
                                    });
                            return stringDeferred;
                        } else {
                            throw new Exception("Table doesn't exists");
                        }

                    }
                }, (Callback<Object, Exception>) e -> {
                    System.out.println(e.getMessage());
                    return e;
                });
            }
        });

        //executing the callback
        res.callback(kuduClient.tableExists(tableName));
    }


A little of the background of my project.  The clients read and write on other Database, and
when they write something, the same information is sent to Kudu. I don't want to block the
client with the Kudu part, because the client only needs the other DataBase to work.

José Ribeiro

________________________________
De: Dan Burkert <danburkert@apache.org>
Enviado: 9 de abril de 2018 17:21
Para: user@kudu.apache.org
Assunto: Re: AsyncKudu

Hi José,

The Deferred class is indeed pretty difficult to come to grips with, which is why we don't
really recommend the async API for most use cases.  I've personally found the Deferred<https://github.com/OpenTSDB/async/blob/master/src/Deferred.java>
class docs to be pretty useful when getting up to speed with it.  Using the AsyncKuduSession
is particularly tricky because it doesn't expose backpressure in an intuitive way, and it
becomes very easy to accidentally have a huge number of writes buffered in the client heap
(the sync client solves this by simply blocking so more rows can't be produced).

kudu-ts<https://github.com/danburkert/kudu-ts/blob/master/core/src/main/java/org/kududb/ts/core/KuduTS.java>
has a little bit of code that uses the async client to open and scan tables, so it may be
helpful to look over that.  It doesn't use the async session, I think because it was easier
just to use the sync client for writing.

- Dan

On Mon, Apr 9, 2018 at 3:10 AM, José Ribeiro <jose.enes.ribeiro@hotmail.com<mailto:jose.enes.ribeiro@hotmail.com>>
wrote:

Hello, I'm having difficulties in the AsyncKudu API. Since there isn't any valuable examples
with asynchronous kudu client, can you please help me in this example?


I want to use AsyncKudu to write in the kudu database. The steps i want to use are:

  1.  verify that the table exists;
  2.  if exists, open table;
  3.  write on the table;


Can someone explain me how to do this with AsyncKudu? I'm having trouble in understanding
the Deferred class that is always returned in every asyncKudu method. Thank you.


Mime
View raw message