Hmm, could you try in spark local mode? i.e. https://jaceklaskowski.
gitbooks.io/mastering-apache-spark/content/spark-local.html
Mike
On Tue, Mar 6, 2018 at 7:14 PM, Ravi Kanth <ravikanth....@gmail.com> wrote:
> Mike,
>
> Can you clarify a bit on grabbing the jstack for the process? I launched
> my Spark application and tried to get the pid using which I thought I can
> grab jstack trace during hang. Unfortunately, I am not able to figure out
> grabbing pid for Spark application.
>
> Thanks,
> Ravi
>
> On 6 March 2018 at 18:36, Mike Percy <mpe...@apache.org> wrote:
>
>> Thanks Ravi. Would you mind attaching the output of jstack on the process
>> during this hang? That would show what the Kudu client threads are doing,
>> as what we are seeing here is just the netty boss thread.
>>
>> Mike
>>
>> On Tue, Mar 6, 2018 at 8:52 AM, Ravi Kanth <ravikanth....@gmail.com>
>> wrote:
>>
>>>
>>> Yes, I have debugged to find the root cause. Every logger before "table
>>> = client.openTable(tableName);" is executing fine and exactly at the
>>> point of opening the table, it is throwing the below exception and nothing
>>> is being executed after that. Still the Spark batches are being processed
>>> and at opening the table is failing. I tried catching it with no luck.
>>> Please find below the exception.
>>>
>>> 8/02/23 00:16:30 ERROR client.TabletClient: [Peer
>>> bd91f34d456a4eccaae50003c90f0fb2] Unexpected exception from downstream
>>> on [id: 0x6e13b01f]
>>> java.net.ConnectException: Connection refused:
>>> kudu102.dev.sac.int.threatmetrix.com/10.112.3.12:7050
>>> at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
>>> at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl
>>> .java:717)
>>> at org.apache.kudu.client.shaded.org.jboss.netty.channel.socket
>>> .nio.NioClientBoss.connect(NioClientBoss.java:152)
>>> at org.apache.kudu.client.shaded.org.jboss.netty.channel.socket
>>> .nio.NioClientBoss.processSelectedKeys(NioClientBoss.java:105)
>>> at org.apache.kudu.client.shaded.org.jboss.netty.channel.socket
>>> .nio.NioClientBoss.process(NioClientBoss.java:79)
>>> at org.apache.kudu.client.shaded.org.jboss.netty.channel.socket
>>> .nio.AbstractNioSelector.run(AbstractNioSelector.java:337)
>>> at org.apache.kudu.client.shaded.org.jboss.netty.channel.socket
>>> .nio.NioClientBoss.run(NioClientBoss.java:42)
>>> at org.apache.kudu.client.shaded.org.jboss.netty.util.ThreadRen
>>> amingRunnable.run(ThreadRenamingRunnable.java:108)
>>> at org.apache.kudu.client.shaded.org.jboss.netty.util.internal.
>>> DeadLockProofWorker$1.run(DeadLockProofWorker.java:42)
>>> at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPool
>>> Executor.java:1142)
>>> at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoo
>>> lExecutor.java:617)
>>> at java.lang.Thread.run(Thread.java:745)
>>>
>>>
>>> Thanks,
>>> Ravi
>>>
>>> On 5 March 2018 at 23:52, Mike Percy <mpe...@apache.org> wrote:
>>>
>>>> Have you considered checking your session error count or pending errors
>>>> in your while loop every so often? Can you identify where your code is
>>>> hanging when the connection is lost (what line)?
>>>>
>>>> Mike
>>>>
>>>> On Mon, Mar 5, 2018 at 9:08 PM, Ravi Kanth <ravikanth....@gmail.com>
>>>> wrote:
>>>>
>>>>> In addition to my previous comment, I raised a support ticket for this
>>>>> issue with Cloudera and one of the support person mentioned below,
>>>>>
>>>>> *"Thank you for clarifying, The exceptions are logged but not
>>>>> re-thrown to an upper layer, so that explains why the Spark application is
>>>>> not aware of the underlying error."*
>>>>>
>>>>> On 5 March 2018 at 21:02, Ravi Kanth <ravikanth....@gmail.com> wrote:
>>>>>
>>>>>> Mike,
>>>>>>
>>>>>> Thanks for the information. But, once the connection to any of the
>>>>>> Kudu servers is lost then there is no way I can have a control on the
>>>>>> KuduSession object and so with getPendingErrors(). The KuduClient in this
>>>>>> case is becoming a zombie and never returned back till the connection is
>>>>>> properly established. I tried doing all that you have suggested with no
>>>>>> luck. Attaching my KuduClient code.
>>>>>>
>>>>>> package org.dwh.streaming.kudu.sparkkudustreaming;
>>>>>>
>>>>>> import java.util.HashMap;
>>>>>> import java.util.Iterator;
>>>>>> import java.util.Map;
>>>>>> import org.apache.hadoop.util.ShutdownHookManager;
>>>>>> import org.apache.kudu.client.*;
>>>>>> import org.apache.spark.api.java.JavaRDD;
>>>>>> import org.slf4j.Logger;
>>>>>> import org.slf4j.LoggerFactory;
>>>>>> import org.dwh.streaming.kudu.sparkkudustreaming.constants.SpecialN
>>>>>> ullConstants;
>>>>>>
>>>>>> public class KuduProcess {
>>>>>> private static Logger logger = LoggerFactory.getLogger(KuduPr
>>>>>> ocess.class);
>>>>>> private KuduTable table;
>>>>>> private KuduSession session;
>>>>>>
>>>>>> public static void upsertKudu(JavaRDD<Map<String, Object>> rdd,
>>>>>> String host, String tableName) {
>>>>>> rdd.foreachPartition(iterator -> {
>>>>>> RowErrorsAndOverflowStatus errors = upsertOpIterator(iterator,
>>>>>> tableName, host);
>>>>>> int errorCount = errors.getRowErrors().length;
>>>>>> if(errorCount > 0){
>>>>>> throw new RuntimeException("Failed to write " + errorCount + "
>>>>>> messages into Kudu");
>>>>>> }
>>>>>> });
>>>>>> }
>>>>>> private static RowErrorsAndOverflowStatus
>>>>>> upsertOpIterator(Iterator<Map<String, Object>> iter, String
>>>>>> tableName, String host) {
>>>>>> try {
>>>>>> AsyncKuduClient asyncClient = KuduConnection.getAsyncClient(host);
>>>>>> KuduClient client = asyncClient.syncClient();
>>>>>> table = client.openTable(tableName);
>>>>>> session = client.newSession();
>>>>>> session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLU
>>>>>> SH_BACKGROUND);
>>>>>> while (iter.hasNext()) {
>>>>>> upsertOp(iter.next());
>>>>>> }
>>>>>> } catch (KuduException e) {
>>>>>> logger.error("Exception in upsertOpIterator method", e);
>>>>>> }
>>>>>> finally{
>>>>>> try {
>>>>>> session.close();
>>>>>> } catch (KuduException e) {
>>>>>> logger.error("Exception in Connection close", e);
>>>>>> }
>>>>>> }
>>>>>> return session.getPendingErrors(); --------------------->
>>>>>> Once, the connection is lost, this part of the code never gets called and
>>>>>> the Spark job will keep on running and processing the records while
>>>>>> the KuduClient is trying to connect to Kudu. Meanwhile, we are loosing
>>>>>> all
>>>>>> the records.
>>>>>> }
>>>>>> public static void upsertOp(Map<String, Object> formattedMap) {
>>>>>> if (formattedMap.size() != 0) {
>>>>>> try {
>>>>>> Upsert upsert = table.newUpsert();
>>>>>> PartialRow row = upsert.getRow();
>>>>>> for (Map.Entry<String, Object> entry : formattedMap.entrySet()) {
>>>>>> if (entry.getValue().getClass().equals(String.class)) {
>>>>>> if (entry.getValue().equals(SpecialNullConstants.specialStringNull))
>>>>>> row.setNull(entry.getKey());
>>>>>> else
>>>>>> row.addString(entry.getKey(), (String) entry.getValue());
>>>>>> } else if (entry.getValue().getClass().equals(Long.class)) {
>>>>>> if (entry.getValue().equals(SpecialNullConstants.specialLongNull))
>>>>>> row.setNull(entry.getKey());
>>>>>> else
>>>>>> row.addLong(entry.getKey(), (Long) entry.getValue());
>>>>>> } else if (entry.getValue().getClass().equals(Integer.class)) {
>>>>>> if (entry.getValue().equals(SpecialNullConstants.specialIntNull))
>>>>>> row.setNull(entry.getKey());
>>>>>> else
>>>>>> row.addInt(entry.getKey(), (Integer) entry.getValue());
>>>>>> }
>>>>>> }
>>>>>>
>>>>>> session.apply(upsert);
>>>>>> } catch (Exception e) {
>>>>>> logger.error("Exception during upsert:", e);
>>>>>> }
>>>>>> }
>>>>>> }
>>>>>> }
>>>>>> class KuduConnection {
>>>>>> private static Logger logger = LoggerFactory.getLogger(KuduCo
>>>>>> nnection.class);
>>>>>> private static Map<String, AsyncKuduClient> asyncCache = new
>>>>>> HashMap<>();
>>>>>> private static int ShutdownHookPriority = 100;
>>>>>>
>>>>>> static AsyncKuduClient getAsyncClient(String kuduMaster) {
>>>>>> if (!asyncCache.containsKey(kuduMaster)) {
>>>>>> AsyncKuduClient asyncClient = new AsyncKuduClient.AsyncKuduClien
>>>>>> tBuilder(kuduMaster).build();
>>>>>> ShutdownHookManager.get().addShutdownHook(new Runnable() {
>>>>>> @Override
>>>>>> public void run() {
>>>>>> try {
>>>>>> asyncClient.close();
>>>>>> } catch (Exception e) {
>>>>>> logger.error("Exception closing async client", e);
>>>>>> }
>>>>>> }
>>>>>> }, ShutdownHookPriority);
>>>>>> asyncCache.put(kuduMaster, asyncClient);
>>>>>> }
>>>>>> return asyncCache.get(kuduMaster);
>>>>>> }
>>>>>> }
>>>>>>
>>>>>>
>>>>>>
>>>>>> Thanks,
>>>>>> Ravi
>>>>>>
>>>>>> On 5 March 2018 at 16:20, Mike Percy <mpe...@apache.org> wrote:
>>>>>>
>>>>>>> Hi Ravi, it would be helpful if you could attach what you are
>>>>>>> getting back from getPendingErrors() -- perhaps from dumping
>>>>>>> RowError.toString() from items in the returned array -- and indicate
>>>>>>> what
>>>>>>> you were hoping to get back. Note that a RowError can also return to you
>>>>>>> the Operation
>>>>>>> <https://kudu.apache.org/releases/1.6.0/apidocs/org/apache/kudu/client/RowError.html#getOperation-->
>>>>>>> that you used to generate the write. From the Operation, you can get the
>>>>>>> original PartialRow
>>>>>>> <https://kudu.apache.org/releases/1.6.0/apidocs/org/apache/kudu/client/PartialRow.html>
>>>>>>> object, which should be able to identify the affected row that the write
>>>>>>> failed for. Does that help?
>>>>>>>
>>>>>>> Since you are using the Kudu client directly, Spark is not involved
>>>>>>> from the Kudu perspective, so you will need to deal with Spark on your
>>>>>>> own
>>>>>>> in that case.
>>>>>>>
>>>>>>> Mike
>>>>>>>
>>>>>>> On Mon, Mar 5, 2018 at 1:59 PM, Ravi Kanth <ravikanth....@gmail.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> Hi Mike,
>>>>>>>>
>>>>>>>> Thanks for the reply. Yes, I am using AUTO_FLUSH_BACKGROUND.
>>>>>>>>
>>>>>>>> So, I am trying to use Kudu Client API to perform UPSERT into Kudu
>>>>>>>> and I integrated this with Spark. I am trying to test a case where in
>>>>>>>> if
>>>>>>>> any of Kudu server fails. So, in this case, if there is any problem in
>>>>>>>> writing, getPendingErrors() should give me a way to handle these
>>>>>>>> errors so
>>>>>>>> that I can successfully terminate my Spark Job. This is what I am
>>>>>>>> trying to
>>>>>>>> do.
>>>>>>>>
>>>>>>>> But, I am not able to get a hold of the exceptions being thrown
>>>>>>>> from with in the KuduClient when retrying to connect to Tablet Server.
>>>>>>>> My
>>>>>>>> getPendingErrors is not getting ahold of these exceptions.
>>>>>>>>
>>>>>>>> Let me know if you need more clarification. I can post some
>>>>>>>> Snippets.
>>>>>>>>
>>>>>>>> Thanks,
>>>>>>>> Ravi
>>>>>>>>
>>>>>>>> On 5 March 2018 at 13:18, Mike Percy <mpe...@apache.org> wrote:
>>>>>>>>
>>>>>>>>> Hi Ravi, are you using AUTO_FLUSH_BACKGROUND
>>>>>>>>> <https://kudu.apache.org/releases/1.6.0/apidocs/org/apache/kudu/client/SessionConfiguration.FlushMode.html>?
>>>>>>>>> You mention that you are trying to use getPendingErrors()
>>>>>>>>> <https://kudu.apache.org/releases/1.6.0/apidocs/org/apache/kudu/client/KuduSession.html#getPendingErrors-->
>>>>>>>>> but
>>>>>>>>> it sounds like it's not working for you -- can you be more specific
>>>>>>>>> about
>>>>>>>>> what you expect and what you are observing?
>>>>>>>>>
>>>>>>>>> Thanks,
>>>>>>>>> Mike
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> On Mon, Feb 26, 2018 at 8:04 PM, Ravi Kanth <
>>>>>>>>> ravikanth....@gmail.com> wrote:
>>>>>>>>>
>>>>>>>>>> Thank Clifford. We are running Kudu 1.4 version. Till date we
>>>>>>>>>> didn't see any issues in production and we are not losing tablet
>>>>>>>>>> servers.
>>>>>>>>>> But, as part of testing I have to generate few unforeseen cases to
>>>>>>>>>> analyse
>>>>>>>>>> the application performance. One among that is bringing down the
>>>>>>>>>> tablet
>>>>>>>>>> server or master server intentionally during which I observed the
>>>>>>>>>> loss of
>>>>>>>>>> records. Just wanted to test cases out of the happy path here. Once
>>>>>>>>>> again
>>>>>>>>>> thanks for taking time to respond to me.
>>>>>>>>>>
>>>>>>>>>> - Ravi
>>>>>>>>>>
>>>>>>>>>> On 26 February 2018 at 19:58, Clifford Resnick <
>>>>>>>>>> cresn...@mediamath.com> wrote:
>>>>>>>>>>
>>>>>>>>>>> I'll have to get back to you on the code bits, but I'm pretty
>>>>>>>>>>> sure we're doing simple sync batching. We're not in production yet,
>>>>>>>>>>> but
>>>>>>>>>>> after some months of development I haven't seen any failures, even
>>>>>>>>>>> when
>>>>>>>>>>> pushing load doing multiple years' backfill. I think the real
>>>>>>>>>>> question is
>>>>>>>>>>> why are you losing tablet servers? The only instability we ever had
>>>>>>>>>>> with
>>>>>>>>>>> Kudu was when it had that weird ntp sync issue that was fixed I
>>>>>>>>>>> think for
>>>>>>>>>>> 1.6. What version are you running?
>>>>>>>>>>>
>>>>>>>>>>> Anyway I would think that infinite loop should be catchable
>>>>>>>>>>> somewhere. Our pipeline is set to fail/retry with Flink snapshots. I
>>>>>>>>>>> imagine there is similar with Spark. Sorry I cant be of more help!
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> On Feb 26, 2018 9:10 PM, Ravi Kanth <ravikanth....@gmail.com>
>>>>>>>>>>> wrote:
>>>>>>>>>>>
>>>>>>>>>>> Cliff,
>>>>>>>>>>>
>>>>>>>>>>> Thanks for the response. Well, I do agree that its simple and
>>>>>>>>>>> seamless. In my case, I am able to upsert ~25000 events/sec into
>>>>>>>>>>> Kudu. But,
>>>>>>>>>>> I am facing the problem when any of the Kudu Tablet or master
>>>>>>>>>>> server is
>>>>>>>>>>> down. I am not able to get a hold of the exception from client. The
>>>>>>>>>>> client
>>>>>>>>>>> is going into an infinite loop trying to connect to Kudu.
>>>>>>>>>>> Meanwhile, I am
>>>>>>>>>>> loosing my records. I tried handling the errors through
>>>>>>>>>>> getPendingErrors()
>>>>>>>>>>> but still it is helpless. I am using AsyncKuduClient to establish
>>>>>>>>>>> the
>>>>>>>>>>> connection and retrieving the syncClient from the Async to open the
>>>>>>>>>>> session
>>>>>>>>>>> and table. Any help?
>>>>>>>>>>>
>>>>>>>>>>> Thanks,
>>>>>>>>>>> Ravi
>>>>>>>>>>>
>>>>>>>>>>> On 26 February 2018 at 18:00, Cliff Resnick <cre...@gmail.com>
>>>>>>>>>>> wrote:
>>>>>>>>>>>
>>>>>>>>>>> While I can't speak for Spark, we do use the client API from
>>>>>>>>>>> Flink streaming and it's simple and seamless. It's especially nice
>>>>>>>>>>> if you
>>>>>>>>>>> require an Upsert semantic.
>>>>>>>>>>>
>>>>>>>>>>> On Feb 26, 2018 7:51 PM, "Ravi Kanth" <ravikanth....@gmail.com>
>>>>>>>>>>> wrote:
>>>>>>>>>>>
>>>>>>>>>>> Hi,
>>>>>>>>>>>
>>>>>>>>>>> Anyone using Spark Streaming to ingest data into Kudu and using
>>>>>>>>>>> Kudu Client API to do so rather than the traditional KuduContext
>>>>>>>>>>> API? I am
>>>>>>>>>>> stuck at a point and couldn't find a solution.
>>>>>>>>>>>
>>>>>>>>>>> Thanks,
>>>>>>>>>>> Ravi
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>
