title: how to connect to Cloud SQL from Google DataFlow
body: <p>I'm trying to create a pipeline task using beam java SDK and Google Dataflow to move data from Cloud SQL to Elastic search</p>
  <p>I've created the following class main method:</p>
  <pre class="lang-java prettyprint-override"><code> public static void main(String[] args) throws Exception{
      DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
      options.setProject(&quot;staging&quot;);      
      options.setTempLocation(&quot;gs://csv_to_sql_staging/temp&quot;); 
      options.setRunner(DataflowRunner.class);        
      options.setGcpTempLocation(&quot;gs://csv_to_sql_staging/temp&quot;);
      options.setUsePublicIps(false);         
      options.setJobName(&quot;tamer-new&quot;);
      options.setSubnetwork(&quot;regions/us-central1/subnetworks/new-network&quot;);
      final List&lt;String&gt; SCOPES = Arrays.asList(
        &quot;https://www.googleapis.com/auth/cloud-platform&quot;,
        &quot;https://www.googleapis.com/auth/devstorage.full_control&quot;,
        &quot;https://www.googleapis.com/auth/userinfo.email&quot;,
        &quot;https://www.googleapis.com/auth/datastore&quot;,
        &quot;https://www.googleapis.com/auth/sqlservice.admin&quot;,
        &quot;https://www.googleapis.com/auth/pubsub&quot;);
      options.setGcpCredential(ServiceAccountCredentials.fromStream(new ElasticSearchIO().getClass().getResourceAsStream(&quot;/staging-b648da5d2b9b.json&quot;)).createScoped(SCOPES));            options.setServiceAccount(&quot;data-flow@staging.iam.gserviceaccount.com&quot;);
                      
      Pipeline p = Pipeline.create(options);
                        
      p.begin();
                        
      PCollection &lt; List &lt; String &gt;&gt; rows = p.apply(JdbcIO. &lt; List &lt; String &gt;&gt; read().withQuery(&quot;select u.id, u.name from user_table&quot;).withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(&quot;com.mysql.jdbc.Driver&quot;, &quot;jdbc:mysql://google/nameDB_new?cloudSqlInstance=staging:europe-west1:sql-staging-instance&amp;socketFactory=com.google.cloud.sql.mysql.SocketFactory&amp;useUnicode=true&amp;characterEncoding=UTF-8&amp;user=user&amp;password=password&amp;useSSL=false&quot;)).withRowMapper(new RowMapper &lt; List &lt; String &gt;&gt; () {
  
          @Override public List &lt; String &gt; mapRow(ResultSet resultSet) throws Exception {
              List &lt; String &gt; addRow = new ArrayList &lt; String &gt; ();
  
  
              for (int i = 1; i &lt;= resultSet.getMetaData().getColumnCount(); i++) {
                  addRow.add(i - 1, String.valueOf(resultSet.getObject(i)));
              }
  
              //LOG.info(String.join(&quot;,&quot;, addRow));
  
              return addRow;
  
          }
      })
  
      .withCoder(ListCoder.of(StringUtf8Coder. &lt; Object &gt; of ()))
  
      );
  
      Write w = ElasticsearchIO.write().withConnectionConfiguration(
          ElasticsearchIO.ConnectionConfiguration.create(new String[] {
              &quot;https://host:9243&quot;
          }, &quot;user-temp&quot;, &quot;String&quot;).withUsername(&quot;elastic&quot;).withPassword(&quot;password&quot;)
      );
  
      rows.apply(w.compose(new SerializableFunction() {
  
         @Override public Object apply(Object input) {
           // TODO Auto-generated method stub
           return input;
          }
      }));
  
  
      p.run().waitUntilFinish();
  
  }
  </code></pre>
  <p>and below is the pom.xml file :</p>
  <pre><code>&lt;?xml version=&quot;1.0&quot; encoding=&quot;UTF-8&quot;?&gt;
  
  &lt;project xmlns=&quot;http://maven.apache.org/POM/4.0.0&quot;
      xmlns:xsi=&quot;http://www.w3.org/2001/XMLSchema-instance&quot;
      xsi:schemaLocation=&quot;http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd&quot;&gt;
    &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
  
    &lt;groupId&gt;com.harmonica.dataflow&lt;/groupId&gt;
    &lt;artifactId&gt;com-harmonica-dataflow&lt;/artifactId&gt;
    &lt;version&gt;0.0.1-SNAPSHOT&lt;/version&gt;
  
    &lt;properties&gt;
      &lt;project.build.sourceEncoding&gt;UTF-8&lt;/project.build.sourceEncoding&gt;
      &lt;maven-compiler-plugin.version&gt;3.7.0&lt;/maven-compiler-plugin.version&gt;
      &lt;exec-maven-plugin.version&gt;1.6.0&lt;/exec-maven-plugin.version&gt;
      &lt;slf4j.version&gt;1.7.25&lt;/slf4j.version&gt;
      &lt;beam.version&gt;2.19.0&lt;/beam.version&gt;
    &lt;/properties&gt;
  
    &lt;repositories&gt;
      &lt;repository&gt;
        &lt;id&gt;ossrh.snapshots&lt;/id&gt;
        &lt;name&gt;Sonatype OSS Repository Hosting&lt;/name&gt;
        &lt;url&gt;https://oss.sonatype.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;${exec-maven-plugin.version}&lt;/version&gt;
            &lt;configuration&gt;
              &lt;cleanupDaemonThreads&gt;false&lt;/cleanupDaemonThreads&gt;
            &lt;/configuration&gt;
          &lt;/plugin&gt;
        &lt;/plugins&gt;
      &lt;/pluginManagement&gt;
    &lt;/build&gt;
  
    &lt;dependencies&gt;
  &lt;!--  Beam Lib --&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;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;dependency&gt;
      &lt;groupId&gt;org.apache.beam&lt;/groupId&gt;
      &lt;artifactId&gt;beam-sdks-java-io-elasticsearch&lt;/artifactId&gt;
      &lt;version&gt;${beam.version}&lt;/version&gt;  
      &lt;/dependency&gt;
  
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.beam&lt;/groupId&gt;
          &lt;artifactId&gt;beam-sdks-java-io-jdbc&lt;/artifactId&gt;
          &lt;version&gt;${beam.version}&lt;/version&gt;
      &lt;/dependency&gt;
      
      
      
      &lt;dependency&gt;
        &lt;groupId&gt;org.apache.beam&lt;/groupId&gt;
        &lt;artifactId&gt;beam-sdks-java-io-google-cloud-platform&lt;/artifactId&gt;
        &lt;version&gt;${beam.version}&lt;/version&gt;
        
      &lt;/dependency&gt;
  
  
          &lt;dependency&gt;
              &lt;groupId&gt;mysql&lt;/groupId&gt;
              &lt;artifactId&gt;mysql-connector-java&lt;/artifactId&gt;
              &lt;version&gt;8.0.19&lt;/version&gt;           
          &lt;/dependency&gt;
  
  &lt;dependency&gt;
      &lt;groupId&gt;com.google.cloud.sql&lt;/groupId&gt;
      &lt;artifactId&gt;mysql-socket-factory-connector-j-8&lt;/artifactId&gt;
      &lt;version&gt;1.0.15&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;/dependencies&gt;
  &lt;/project&gt;
  </code></pre>
  <p>and when I execute this command:</p>
  <pre><code>man exec mvn compile exec:java -Dexec.mainClass=com.dataflow.ElasticSearchIO 
  </code></pre>
  <p>The worker started successfully but then cant connect to the the Cloud SQL:
  even thought I've done the flowing:</p>
  <ul>
  <li>I've created a service account with owner access to the project and passed it to the runner options</li>
  <li>I've created a VPC network with a name of new-network with a range of IP of 190.10.0.0/16 and assigned it to the pipeline options and then whitlisted this range in the cloud SQL</li>
  </ul>
  <p>and however Im still getting this error:</p>
  <blockquote>
  <p>Error message from worker: java.lang.RuntimeException:
  org.apache.beam.sdk.util.UserCodeException: java.sql.SQLException:
  Cannot create PoolableConnectionFactory (Communications link failure
  The last packet sent successfully to the server was 0 milliseconds
  ago. The driver has not received any packets from the server.)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$1.typedApply(IntrinsicMapTaskExecutorFactory.java:194)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$1.typedApply(IntrinsicMapTaskExecutorFactory.java:165)
  org.apache.beam.runners.dataflow.worker.graph.Networks$TypeSafeNodeFunction.apply(Networks.java:63)
  org.apache.beam.runners.dataflow.worker.graph.Networks$TypeSafeNodeFunction.apply(Networks.java:50)
  org.apache.beam.runners.dataflow.worker.graph.Networks.replaceDirectedNetworkNodes(Networks.java:87)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory.create(IntrinsicMapTaskExecutorFactory.java:125)
  org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.doWork(BatchDataflowWorker.java:352)
  org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.getAndPerformWork(BatchDataflowWorker.java:305)
  org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.doWork(DataflowBatchWorkerHarness.java:140)
  org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.call(DataflowBatchWorkerHarness.java:120)
  org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.call(DataflowBatchWorkerHarness.java:107)
  java.util.concurrent.FutureTask.run(FutureTask.java:266)
  java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  java.lang.Thread.run(Thread.java:748) Caused by:
  org.apache.beam.sdk.util.UserCodeException: java.sql.SQLException:
  Cannot create PoolableConnectionFactory (Communications link failure
  The last packet sent successfully to the server was 0 milliseconds
  ago. The driver has not received any packets from the server.)
  org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:34)
  org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn$DoFnInvoker.invokeSetup(Unknown
  Source)
  org.apache.beam.runners.dataflow.worker.DoFnInstanceManagers$ConcurrentQueueInstanceManager.deserializeCopy(DoFnInstanceManagers.java:80)
  org.apache.beam.runners.dataflow.worker.DoFnInstanceManagers$ConcurrentQueueInstanceManager.peek(DoFnInstanceManagers.java:62)
  org.apache.beam.runners.dataflow.worker.UserParDoFnFactory.create(UserParDoFnFactory.java:95)
  org.apache.beam.runners.dataflow.worker.DefaultParDoFnFactory.create(DefaultParDoFnFactory.java:75)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory.createParDoOperation(IntrinsicMapTaskExecutorFactory.java:264)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory.access$000(IntrinsicMapTaskExecutorFactory.java:86)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$1.typedApply(IntrinsicMapTaskExecutorFactory.java:183)
  ... 14 more Caused by: java.sql.SQLException: Cannot create
  PoolableConnectionFactory (Communications link failure The last packet
  sent successfully to the server was 0 milliseconds ago. The driver has
  not received any packets from the server.)
  org.apache.commons.dbcp2.BasicDataSource.createPoolableConnectionFactory(BasicDataSource.java:735)
  org.apache.commons.dbcp2.BasicDataSource.createDataSource(BasicDataSource.java:605)
  org.apache.commons.dbcp2.BasicDataSource.getConnection(BasicDataSource.java:809)
  org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn.setup(JdbcIO.java:881)
  Caused by: com.mysql.cj.jdbc.exceptions.CommunicationsException:
  Communications link failure The last packet sent successfully to the
  server was 0 milliseconds ago. The driver has not received any packets
  from the server.
  com.mysql.cj.jdbc.exceptions.SQLError.createCommunicationsException(SQLError.java:174)
  com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:64)
  com.mysql.cj.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:836)
  com.mysql.cj.jdbc.ConnectionImpl.(ConnectionImpl.java:456)
  com.mysql.cj.jdbc.ConnectionImpl.getInstance(ConnectionImpl.java:246)
  com.mysql.cj.jdbc.NonRegisteringDriver.connect(NonRegisteringDriver.java:197)
  org.apache.commons.dbcp2.DriverConnectionFactory.createConnection(DriverConnectionFactory.java:53)
  org.apache.commons.dbcp2.PoolableConnectionFactory.makeObject(PoolableConnectionFactory.java:355)
  org.apache.commons.dbcp2.BasicDataSource.validateConnectionFactory(BasicDataSource.java:116)
  org.apache.commons.dbcp2.BasicDataSource.createPoolableConnectionFactory(BasicDataSource.java:731)
  org.apache.commons.dbcp2.BasicDataSource.createDataSource(BasicDataSource.java:605)
  org.apache.commons.dbcp2.BasicDataSource.getConnection(BasicDataSource.java:809)
  org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn.setup(JdbcIO.java:881)
  org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn$DoFnInvoker.invokeSetup(Unknown
  Source)
  org.apache.beam.runners.dataflow.worker.DoFnInstanceManagers$ConcurrentQueueInstanceManager.deserializeCopy(DoFnInstanceManagers.java:80)
  org.apache.beam.runners.dataflow.worker.DoFnInstanceManagers$ConcurrentQueueInstanceManager.peek(DoFnInstanceManagers.java:62)
  org.apache.beam.runners.dataflow.worker.UserParDoFnFactory.create(UserParDoFnFactory.java:95)
  org.apache.beam.runners.dataflow.worker.DefaultParDoFnFactory.create(DefaultParDoFnFactory.java:75)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory.createParDoOperation(IntrinsicMapTaskExecutorFactory.java:264)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory.access$000(IntrinsicMapTaskExecutorFactory.java:86)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$1.typedApply(IntrinsicMapTaskExecutorFactory.java:183)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$1.typedApply(IntrinsicMapTaskExecutorFactory.java:165)
  org.apache.beam.runners.dataflow.worker.graph.Networks$TypeSafeNodeFunction.apply(Networks.java:63)
  org.apache.beam.runners.dataflow.worker.graph.Networks$TypeSafeNodeFunction.apply(Networks.java:50)
  org.apache.beam.runners.dataflow.worker.graph.Networks.replaceDirectedNetworkNodes(Networks.java:87)
  org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory.create(IntrinsicMapTaskExecutorFactory.java:125)
  org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.doWork(BatchDataflowWorker.java:352)
  org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.getAndPerformWork(BatchDataflowWorker.java:305)
  org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.doWork(DataflowBatchWorkerHarness.java:140)
  org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.call(DataflowBatchWorkerHarness.java:120)
  org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.call(DataflowBatchWorkerHarness.java:107)
  java.util.concurrent.FutureTask.run(FutureTask.java:266)
  java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  java.lang.Thread.run(Thread.java:748) Caused by:
  com.mysql.cj.exceptions.CJCommunicationsException: Communications link
  failure The last packet sent successfully to the server was 0
  milliseconds ago. The driver has not received any packets from the
  server. sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native
  Method)
  sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
  sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
  java.lang.reflect.Constructor.newInstance(Constructor.java:423)
  com.mysql.cj.exceptions.ExceptionFactory.createException(ExceptionFactory.java:61)
  com.mysql.cj.exceptions.ExceptionFactory.createException(ExceptionFactory.java:105)
  com.mysql.cj.exceptions.ExceptionFactory.createException(ExceptionFactory.java:151)
  com.mysql.cj.exceptions.ExceptionFactory.createCommunicationsException(ExceptionFactory.java:167)
  com.mysql.cj.protocol.a.NativeSocketConnection.connect(NativeSocketConnection.java:91)
  com.mysql.cj.NativeSession.connect(NativeSession.java:144)
  com.mysql.cj.jdbc.ConnectionImpl.connectOneTryOnly(ConnectionImpl.java:956)
  com.mysql.cj.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:826)
  ... 32 more Caused by: java.net.ConnectException: Connection timed out
  (Connection timed out) java.net.PlainSocketImpl.socketConnect(Native
  Method)
  java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
  java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
  java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
  java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
  java.net.Socket.connect(Socket.java:589)
  sun.security.ssl.SSLSocketImpl.connect(SSLSocketImpl.java:673)
  sun.security.ssl.BaseSSLSocketImpl.connect(BaseSSLSocketImpl.java:173)
  com.google.cloud.sql.core.CoreSocketFactory.createSslSocket(CoreSocketFactory.java:233)
  com.google.cloud.sql.core.CoreSocketFactory.connect(CoreSocketFactory.java:185)
  com.google.cloud.sql.mysql.SocketFactory.connect(SocketFactory.java:48)
  com.google.cloud.sql.mysql.SocketFactory.connect(SocketFactory.java:38)
  com.mysql.cj.protocol.a.NativeSocketConnection.connect(NativeSocketConnection.java:65)
  ... 35 more</p>
  </blockquote>
  <p>Plz any help will be highly appreciated!
  Thanks in advance</p>