title: Apache Beam for Google Cloud DataFlow - 404 errors when using BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED
body: <p>I'm experiencing an issue when I run a Beam DataFlow (written in Java), for data entry into Google BigQuery. The issue occurs in my development environment and also on the production environment.</p>
  
  <p>I'm processing multiple items of data through a Beam pipeline on separate threads. I'm processing by using a <code>ParDo</code> that transforms the received data using an <code>apply</code> over a <code>PCollection</code>. For the transformed data, I then attempt to write it to Google BigQuery.</p>
  
  <p>I'm attempting to write my transformed data into BigQuery with a routine that looks like the following: </p>
  
  <pre><code>transformedData
     .apply("Load fact data",
           BigQueryIO.&lt;ValidatedDataRecord&gt;write()
           .to(new LoadDataFact.DynamicFactTableDestination(dataType.label))
           .withFormatFunction(new LoadDataFact.FactSerializationFn())
           .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
           .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
  </code></pre>
  
  <p>The significant thing here is that I'm using the following:</p>
  
  <pre><code>.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED
  </code></pre>
  
  <p>... because I want a new table to be created if doesn't already exist</p>
  
  <p>The issue is that when new data becomes part of the dataflow, and that data requires a new dataset for the data, I frequently (but don't always) get the following 404 error:</p>
  
  <pre><code>Exception thrown in class : com.myOrg.myPackage.myClass Error : java.lang.RuntimeException: com.google.api.client.googleapis.json.GoogleJsonResponseException: 404 Not Found
  {
  "code" : 404,
  "errors" : [ {
  "domain" : "global",
  "message" : "Not found: Table my-project:my-dataset.my-table",
  "reason" : "notFound"
  } ],
  "message" : "Not found: Table my-project:my-dataset.my-table",
  "status" : "NOT_FOUND"
  }
  </code></pre>
  
  <p>I've investigated what goes on in the underlying Beam SDKs Java Core library. What's happening is that when the 1st new item of data is processed, the <code>CreateTables.class</code> in the 'Beam SDKs Java Core' library should try to create a new table, and if successful it adds an entry to the static <code>CreateTables.createdTables</code> collection for the new table, to indicate that the table is created. The 404 error seems to occur subsequent to the table being created (although sometimes, the table fails entirely to be created). I'm not sure what's causing this behaviour, or what triggers the 404 error (the error message doesn't give much info). This type of error in BigQuery is usually because the table doesn't exist at whatever stage it's being accessed.</p>
  
  <p>The <code>CreateTables.class</code> is part of the following Beam SDKs Java Core library:</p>
  
  <pre><code>C:\Users\my.username\.m2\repository\org\apache\beam\beam-sdks-java-io-google-cloud-platform\2.5.0\beam-sdks-java-io-google-cloud-platform-2.5.0.jar!\org\apache\beam\sdk\io\gcp\bigquery\CreateTables.class
  </code></pre>
  
  <p>I've raised this a a bug on the Beam Jira - <a href="https://issues.apache.org/jira/browse/BEAM-7195" rel="nofollow noreferrer">https://issues.apache.org/jira/browse/BEAM-7195</a></p>
  
  <p>I tried updating my Beam SDKs Java Core library to v 2.12.0, but for some reason the dataflow stopped working altogether. </p>
  
  <p>Is there a way I can work around the issue? I'm using a custom class to override the <code>DynamicDestinations</code> class that Beam uses for processing (this is the way that the documentation specifies that dynamic data destinations should be implemented - see <a href="https://beam.apache.org/documentation/io/built-in/google-bigquery/#using-dynamic-destinations" rel="nofollow noreferrer">https://beam.apache.org/documentation/io/built-in/google-bigquery/#using-dynamic-destinations</a>).</p>
  
  <p>My custom class looks like this, and the <code>getTable</code> method is called during the process of each item of data, to determine the table that the data should be added to:</p>
  
  <pre><code>public class LoadDataFact {
  
      public static class DynamicFactTableDestination extends DynamicDestinations&lt;ValidatedDataRecord, String&gt; {
          private static final long serialVersionUID = -1234561111111123456L;
          private static final String projectID = "my-project";
          private String dataType = "none";
          private String elementDuration = "unknown";
  
          public DynamicFactTableDestination(String dataType) {
              this.dataType = dataType;
          }
  
          @Override
          public String getDestination(ValueInSingleWindow&lt;ValidatedDataRecord&gt; element) {    
  
              return element.getValue().DatasetName;
          }
  
          @Override
          public TableDestination getTable(String destination) {
  
              try {
                  return new TableDestination(new TableReference()
                          .setProjectId(projectID)
                          .setDatasetId(destination)
                          .setTableId(String.format("data_for_%s",this.dataType)), "Data staging table",
                          new TimePartitioning()
                                  .setType("DAY")
                                  .setField("created_date_time"));
              }
              catch (Exception ex) {
                  System.out.println("Error " + ex.getMessage());
              }
  
              return null;
          }
  </code></pre>
  
  <p>Here's my pom.xml:</p>
  
  <pre><code>&lt;?xml version="1.0" encoding="UTF-8"?&gt;
  &lt;!--
      Licensed to the Apache Software Foundation (ASF) under one or more
      contributor license agreements.  See the NOTICE file distributed with
      this work for additional information regarding copyright ownership.
      The ASF licenses this file to You under the Apache License, Version 2.0
      (the "License"); you may not use this file except in compliance with
      the License.  You may obtain a copy of the License at
  
         http://www.apache.org/licenses/LICENSE-2.0
  
      Unless required by applicable law or agreed to in writing, software
      distributed under the License is distributed on an "AS IS" BASIS,
      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
      See the License for the specific language governing permissions and
      limitations under the License.
  --&gt;
  &lt;project xmlns="http://maven.apache.org/POM/4.0.0"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
    &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
  
    &lt;groupId&gt;com.myOrg.myNamespace&lt;/groupId&gt;
    &lt;artifactId&gt;my-artifact-name&lt;/artifactId&gt;
    &lt;version&gt;1.0-SNAPSHOT&lt;/version&gt;
  
    &lt;packaging&gt;jar&lt;/packaging&gt;
  
    &lt;properties&gt;
      &lt;beam.version&gt;2.5.0&lt;/beam.version&gt;
      &lt;maven-compiler-plugin.version&gt;3.6.2&lt;/maven-compiler-plugin.version&gt;
      &lt;maven-exec-plugin.version&gt;1.6.0&lt;/maven-exec-plugin.version&gt;
      &lt;slf4j.version&gt;1.7.25&lt;/slf4j.version&gt;
    &lt;/properties&gt;
  
    &lt;repositories&gt;
      &lt;repository&gt;
        &lt;id&gt;apache.snapshots&lt;/id&gt;
        &lt;name&gt;Apache Development Snapshot Repository&lt;/name&gt;
        &lt;url&gt;https://repository.apache.org/content/repositories/snapshots/&lt;/url&gt;
        &lt;releases&gt;
          &lt;enabled&gt;false&lt;/enabled&gt;
        &lt;/releases&gt;
        &lt;snapshots&gt;
          &lt;enabled&gt;true&lt;/enabled&gt;
        &lt;/snapshots&gt;
      &lt;/repository&gt;
    &lt;/repositories&gt;
  
    &lt;build&gt;
     &lt;plugins&gt;
        &lt;plugin&gt;
          &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
          &lt;artifactId&gt;maven-compiler-plugin&lt;/artifactId&gt;
          &lt;version&gt;${maven-compiler-plugin.version}&lt;/version&gt;
          &lt;configuration&gt;
            &lt;source&gt;1.8&lt;/source&gt;
            &lt;target&gt;1.8&lt;/target&gt;
          &lt;/configuration&gt;
        &lt;/plugin&gt;
      &lt;/plugins&gt;
  
      &lt;pluginManagement&gt;
        &lt;plugins&gt;
          &lt;plugin&gt;
            &lt;groupId&gt;org.codehaus.mojo&lt;/groupId&gt;
            &lt;artifactId&gt;exec-maven-plugin&lt;/artifactId&gt;
            &lt;version&gt;${maven-exec-plugin.version}&lt;/version&gt;
            &lt;configuration&gt;
              &lt;cleanupDaemonThreads&gt;false&lt;/cleanupDaemonThreads&gt;
            &lt;/configuration&gt;
          &lt;/plugin&gt;
          &lt;!--This plugin's configuration is used to store Eclipse m2e settings only. It has no influence on the Maven build itself.--&gt;
          &lt;plugin&gt;
             &lt;groupId&gt;org.eclipse.m2e&lt;/groupId&gt;
             &lt;artifactId&gt;lifecycle-mapping&lt;/artifactId&gt;
             &lt;version&gt;1.0.0&lt;/version&gt;
             &lt;configuration&gt;
                &lt;lifecycleMappingMetadata&gt;
                   &lt;pluginExecutions&gt;
                      &lt;pluginExecution&gt;
                         &lt;pluginExecutionFilter&gt;
                            &lt;groupId&gt;
                               org.apache.maven.plugins
                            &lt;/groupId&gt;
                            &lt;artifactId&gt;
                               maven-compiler-plugin
                            &lt;/artifactId&gt;
                            &lt;versionRange&gt;
                               [@maven-compiler-plugin.version@,)
                            &lt;/versionRange&gt;
                            &lt;goals&gt;
                               &lt;goal&gt;compile&lt;/goal&gt;
                               &lt;goal&gt;testCompile&lt;/goal&gt;
                            &lt;/goals&gt;
                         &lt;/pluginExecutionFilter&gt;
                         &lt;action&gt;
                            &lt;ignore&gt;&lt;/ignore&gt;
                         &lt;/action&gt;
                      &lt;/pluginExecution&gt;
                   &lt;/pluginExecutions&gt;
                &lt;/lifecycleMappingMetadata&gt;
             &lt;/configuration&gt;
          &lt;/plugin&gt;
        &lt;/plugins&gt;
      &lt;/pluginManagement&gt;
    &lt;/build&gt;
  
    &lt;dependencies&gt;
  
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.beam&lt;/groupId&gt;
        &lt;artifactId&gt;beam-sdks-java-core&lt;/artifactId&gt;
        &lt;version&gt;${beam.version}&lt;/version&gt;
      &lt;/dependency&gt;
  
  
       &lt;dependency&gt;
        &lt;groupId&gt;com.google.cloud.dataflow&lt;/groupId&gt;
        &lt;artifactId&gt;google-cloud-dataflow-java-sdk-all&lt;/artifactId&gt;
        &lt;version&gt;2.5.0&lt;/version&gt;
       &lt;/dependency&gt;
  
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.beam&lt;/groupId&gt;
        &lt;artifactId&gt;beam-runners-google-cloud-dataflow-java&lt;/artifactId&gt;
        &lt;version&gt;${beam.version}&lt;/version&gt;
      &lt;/dependency&gt;
  
  
      &lt;!-- slf4j API frontend binding with JUL backend --&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.slf4j&lt;/groupId&gt;
        &lt;artifactId&gt;slf4j-api&lt;/artifactId&gt;
        &lt;version&gt;${slf4j.version}&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.slf4j&lt;/groupId&gt;
        &lt;artifactId&gt;slf4j-jdk14&lt;/artifactId&gt;
        &lt;version&gt;${slf4j.version}&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.msgpack&lt;/groupId&gt;
        &lt;artifactId&gt;msgpack-core&lt;/artifactId&gt;
        &lt;version&gt;0.8.16&lt;/version&gt;
      &lt;/dependency&gt;
    &lt;/dependencies&gt;
  &lt;/project&gt;
  </code></pre>