title: Connect to Kafka with SSL using KafkaIO on Google Dataflow
body: <p>From a server, I was able to connect and get the data out from a remote kafka server topic which has SSL configured. </p>
  
  <p>From GCP, How can I connect to a remote kafka server using Google Dataflow pipeline passing SSL truststore, keystore certificates locations and the Google service account json?</p>
  
  <p>I am using Eclipse plugin for dataflow runner option.</p>
  
  <p>If I point to certificate on GCS, It throws error when certs are pointed to Google storage bucket. </p>
  
  <pre><code>
  Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
  
  Caused by: org.apache.kafka.common.KafkaException:
   java.io.FileNotFoundException: 
  gs:/bucket/folder/truststore-client.jks (No such file or directory)
  </code></pre>
  
  <p>Followed: <a href="https://stackoverflow.com/questions/42726011/truststore-and-google-cloud-dataflow/53549757#53549757">Truststore and Google Cloud Dataflow</a></p>
  
  <p>Updated code pointing SSL truststore, keystore location to local machine's /tmp directory certifcates in case KafkaIO needs to read from file path. It did not throw FileNotFoundError.</p>
  
  <p>Tried running the server Java client code from the GCP account and also using Dataflow - Beam Java pipeline, I get following error. </p>
  
  <pre><code>
  ssl.truststore.location = &lt;LOCAL MACHINE CERTICATE FILE PATH&gt;
      ssl.truststore.password = [hidden]
      ssl.truststore.type = JKS
      value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
  
  org.apache.kafka.common.utils.AppInfoParser$AppInfo &lt;init&gt;
  INFO: Kafka version : 1.0.0
  org.apache.kafka.common.utils.AppInfoParser$AppInfo &lt;init&gt;
  INFO: Kafka commitId : aaa7af6d4a11b29d
  org.apache.kafka.common.network.SslTransportLayer close
  WARNING: Failed to send SSL Close message 
  java.io.IOException: Broken pipe
  
  org.apache.beam.runners.direct.RootProviderRegistry.getInitialInputs(RootProviderRegistry.java:81)
      at org.apache.beam.runners.direct.ExecutorServiceParallelExecutor.start(ExecutorServiceParallelExecutor.java:153)
      at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:205)
      at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:66)
      at org.apache.beam.sdk.Pipeline.run(Pipeline.java:311)
      at org.apache.beam.sdk.Pipeline.run(Pipeline.java:297)
      at 
  
  org.apache.kafka.common.utils.LogContext$KafkaLogger warn
  WARNING: [Consumer clientId=consumer-1, groupId=test-group] Connection to node -2 terminated during authentication. This may indicate that authentication failed due to invalid credentials.
  
  </code></pre>
  
  <p>Any suggestions or examples appreciated.</p>