title: Apache Beam pipeline running on Dataflow failed to read from KafkaIO: SSL handshake failed
body: <p>I'm building an Apache Beam pipeline to read from Kafka as an unbounded source. </p>
  
  <p>I was able to run it locally using direct runner. </p>
  
  <p>However, the pipeline would fail with the attached exception stack trace, when run using Google Cloud Dataflow runner on the cloud.</p>
  
  <p>It seems it's ultimately the <a href="https://github.com/google/conscrypt" rel="noreferrer">Conscrypt Java library</a> that's throwing <a href="https://github.com/google/conscrypt/blob/1.1.x/common/src/main/java/org/conscrypt/ConscryptEngine.java#L782" rel="noreferrer"><code>javax.net.ssl.SSLException: Unable to parse TLS packet header</code></a>. I'm not really sure how to address this issue.</p>
  
  <pre><code>java.io.IOException: Failed to start reading from source: org.apache.beam.sdk.io.kafka.KafkaUnboundedSource@33b5ff70
          com.google.cloud.dataflow.worker.WorkerCustomSources$UnboundedReaderIterator.start(WorkerCustomSources.java:783)
          com.google.cloud.dataflow.worker.util.common.worker.ReadOperation$SynchronizedReaderIterator.start(ReadOperation.java:360)
          com.google.cloud.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:193)
          com.google.cloud.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:158)
          com.google.cloud.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:75)
          com.google.cloud.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1227)
          com.google.cloud.dataflow.worker.StreamingDataflowWorker.access$1000(StreamingDataflowWorker.java:135)
          com.google.cloud.dataflow.worker.StreamingDataflowWorker$6.run(StreamingDataflowWorker.java:966)
          java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
          java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          java.lang.Thread.run(Thread.java:745)
  Caused by: java.io.IOException: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
          org.apache.beam.sdk.io.kafka.KafkaUnboundedReader.start(KafkaUnboundedReader.java:126)
          com.google.cloud.dataflow.worker.WorkerCustomSources$UnboundedReaderIterator.start(WorkerCustomSources.java:778)
          com.google.cloud.dataflow.worker.util.common.worker.ReadOperation$SynchronizedReaderIterator.start(ReadOperation.java:360)
          com.google.cloud.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:193)
          com.google.cloud.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:158)
          com.google.cloud.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:75)
          com.google.cloud.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1227)
          com.google.cloud.dataflow.worker.StreamingDataflowWorker.access$1000(StreamingDataflowWorker.java:135)
          com.google.cloud.dataflow.worker.StreamingDataflowWorker$6.run(StreamingDataflowWorker.java:966)
          java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
          java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          java.lang.Thread.run(Thread.java:745)
  Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
          java.util.concurrent.FutureTask.report(FutureTask.java:122)
          java.util.concurrent.FutureTask.get(FutureTask.java:206)
          org.apache.beam.sdk.io.kafka.KafkaUnboundedReader.start(KafkaUnboundedReader.java:112)
          com.google.cloud.dataflow.worker.WorkerCustomSources$UnboundedReaderIterator.start(WorkerCustomSources.java:778)
          com.google.cloud.dataflow.worker.util.common.worker.ReadOperation$SynchronizedReaderIterator.start(ReadOperation.java:360)
          com.google.cloud.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:193)
          com.google.cloud.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:158)
          com.google.cloud.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:75)
          com.google.cloud.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1227)
          com.google.cloud.dataflow.worker.StreamingDataflowWorker.access$1000(StreamingDataflowWorker.java:135)
          com.google.cloud.dataflow.worker.StreamingDataflowWorker$6.run(StreamingDataflowWorker.java:966)
          java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
          java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          java.lang.Thread.run(Thread.java:745)
  Caused by: org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
  Caused by: javax.net.ssl.SSLException: Unable to parse TLS packet header
          org.conscrypt.ConscryptEngine.unwrap(ConscryptEngine.java:782)
          org.conscrypt.ConscryptEngine.unwrap(ConscryptEngine.java:723)
          org.conscrypt.ConscryptEngine.unwrap(ConscryptEngine.java:688)
          org.conscrypt.Java8EngineWrapper.unwrap(Java8EngineWrapper.java:236)
          org.apache.kafka.common.network.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:464)
          org.apache.kafka.common.network.SslTransportLayer.doHandshake(SslTransportLayer.java:328)
          org.apache.kafka.common.network.SslTransportLayer.handshake(SslTransportLayer.java:255)
          org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:79)
          org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:460)
          org.apache.kafka.common.network.Selector.poll(Selector.java:398)
          org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:460)
          org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:238)
          org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:214)
          org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:190)
          org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:219)
          org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:205)
          org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.fetchCommittedOffsets(ConsumerCoordinator.java:468)
          org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.refreshCommittedOffsetsIfNeeded(ConsumerCoordinator.java:450)
          org.apache.kafka.clients.consumer.KafkaConsumer.updateFetchPositions(KafkaConsumer.java:1772)
          org.apache.kafka.clients.consumer.KafkaConsumer.position(KafkaConsumer.java:1411)
          org.apache.beam.sdk.io.kafka.KafkaUnboundedReader.setupInitialOffset(KafkaUnboundedReader.java:641)
          org.apache.beam.sdk.io.kafka.KafkaUnboundedReader.lambda$start$0(KafkaUnboundedReader.java:106)
          java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
          java.util.concurrent.FutureTask.run(FutureTask.java:266)
          java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
          java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          java.lang.Thread.run(Thread.java:745)
  </code></pre>