NullPointerException while waiting for query result on BigQuery

[Astrac]

I have a complex application and I am using Apache Flink to stream events from Kafka to BigQuery; the application queries the dataset first in order to get the latest offset for each topic/partition. This is how the code running the query looks like:

var response = retryUntilSuccess(maxFetchOffsetAttempts, maxFetchOffsetDelay) { () =>
  Thread.sleep(Random.nextInt(maxFetchOffsetDelay.toMillis.toInt))
  bigQuery.query(
    QueryRequest
      .newBuilder(offsetsQuery)
      .setUseLegacySql(false)
      .setDefaultDataset(config.dataset)
      .build()
  )
}

while (!response.jobCompleted()) {
  logger.warn(
    s"Offsets for ${record.tableName} are not ready, will retry in 1 second, jobId: ${response.getJobId()}")
  Thread.sleep(1000)
  response = bigQuery.getQueryResults(response.getJobId())
}

val result = response.getResult()

This is written in Scala but it is mostly a translation of the example in the Javadocs; the retryUntilSuccess function retries the request multiple time and there is a random wait (30 seconds at this moment) before the request so to handle the fact that many queries will fire in parallel and may result in hitting API limits.

When I run this code I almost always get the following exception:

java.lang.NullPointerException
	at com.google.cloud.bigquery.JobId.fromPb(JobId.java:111)
	at com.google.cloud.bigquery.BigQueryImpl.getQueryResults(BigQueryImpl.java:635)
	at com.google.cloud.bigquery.BigQueryImpl.getQueryResults(BigQueryImpl.java:619)
	at io.chumps.dataprocessing.sinks.BigQuerySink$$anon$1.startingOffsets$lzycompute(BigQuerySink.scala:89)
	at io.chumps.dataprocessing.sinks.BigQuerySink$$anon$1.startingOffsets(BigQuerySink.scala:73)
	at io.chumps.dataprocessing.sinks.BigQuerySink$$anon$1.open(BigQuerySink.scala:136)
	at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:38)
	at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:91)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:376)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:256)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:585)
	at java.lang.Thread.run(Thread.java:745)

The line in our code where this exception happens is where I call getQueryResults and I am sure that I am not sending a null in there as I can see the job id in the logs before the failure; I also checked that the response doesn't contain any error by calling getExecutionErrors. I say that "it almost always fails" because over some tens attempts it managed to run a couple of times without incurring in the exception.

Could this be a bug in the way the API limits are hadled by the client? Do you have any advice to work around this issue?

Thanks!

[vaichidrewar]

I also got the same error.
Caused by: java.lang.NullPointerException
at com.google.cloud.bigquery.JobId.fromPb(JobId.java:97)
at com.google.cloud.bigquery.BigQueryImpl.getQueryResults(BigQueryImpl.java:610)
at com.google.cloud.bigquery.BigQueryImpl.getQueryResults(BigQueryImpl.java:594)
at com.bloomreach.merchandising.storage.BasicDBUtilBigQuery.executeQueryOnBigQuery(BasicDBUtilBigQuery.java:144)

---

public QueryResult executeQueryOnBigQuery(String sql) throws BQQException, InterruptedException {
QueryResponse response;
QueryRequest queryRequest = QueryRequest.newBuilder(sql)
.setUseLegacySql(false)
.build();
try {
long start = System.currentTimeMillis();
response = client.query(queryRequest);
long finish = System.currentTimeMillis();
log.info("executeQueryOnBigQuery running time (ms):" + (finish - start));
} catch (BigQueryException e) {
throw new BQQException(e);
}

while (!response.jobCompleted()) {
  TimeUnit.MILLISECONDS.sleep(DEFAULT_SLEEP_INTERVAL);
  try {
    response = client.getQueryResults(response.getJobId());        << line   144
  } catch (BigQueryException e) {
    throw new BQQException("Failed to grab query results" + e, e);
  }
}

List<BigQueryError> executionErrors = response.getExecutionErrors();
if (!executionErrors.isEmpty()) {
  throw new BQQException("BigQueryError", executionErrors);
}

QueryResult result = response.getResult();
return result;

}

[cobookman]

Added an internal issue w/in Google under bug id #36662731

[cobookman]

It looks like the jobs.getQueryResults method returns a response that contains the "error" field but no "jobReference" field when the query fails or is invalid. This is a non-retryable error that will result in an NullPointerException when dereferencing the jobResponse here. The client library needs to be updated to handle this case. I've talked to the internal Google team and they are working on getting out that fix.

As for workarounds to this issue with the client library today. You could call Job.waitFor() to poll for job completion and then check for errors before calling getQueryResults().

Here is a code snippet using this Job.waitFor() method call:

SimpleApp.java L42-74

    QueryJobConfiguration queryConfig =
        QueryJobConfiguration.newBuilder(
            "SELECT "
                + "APPROX_TOP_COUNT(corpus, 10) as title, "
                + "COUNT(*) as unique_words "
                + "FROM `publicdata.samples.shakespeare`;")
            // Use standard SQL syntax for queries.
            // See: https://cloud.google.com/bigquery/sql-reference/
            .setUseLegacySql(false)
            .build();

    // Create a job ID so that we can safely retry.
    JobId jobId = JobId.of(UUID.randomUUID().toString());
    Job queryJob = bigquery.create(JobInfo.newBuilder(queryConfig).setJobId(jobId).build());

    // Wait for the query to complete.
    queryJob = queryJob.waitFor();

    // Check for errors
    if (queryJob == null) {
      throw new RuntimeException("Job no longer exists");
    } else if (queryJob.getStatus().getError() != null) {
      // You can also look at queryJob.getStatus().getExecutionErrors() for all
      // errors, not just the latest one.
      throw new RuntimeException(queryJob.getStatus().getError().toString());
    }

    // Get the results.
    QueryResponse response = bigquery.getQueryResults(jobId);
    // [END run_query]

    // [START print_results]
    QueryResult result = response.getResult();

[riccardoref]

I am running queries that take about 46 seconds to run. The issue seems less severe when waiting 10 seconds instead of 1.

[saicheems]

So I can see where the issue is...

2 things:

1. What's the expected result? What should the client do if it gets a response with errors but no jobReference field?
2. Is there a query on a public dataset that can reproduce?

[garrettjonesgoogle]

Downgrading to P2 because this issue is hard to reproduce.