PROSAGA码农传奇-可信计算-Truststore和Google Cloud Dataflow

感谢@jkff的解决方案，这是一个实现示例：

ConsumerFactoryFn实现示例：


    
          private static class ConsumerFactoryFn
        implements SerializableFunction, Consumer>
{
public Consumer<byte[], byte[]> apply(Map<String, Object> config) 
{
    try 
    {
        Storage storage = StorageOptions.newBuilder()
                .setProjectId("prj-id-of-your-bucket")
                .setCredentials(GoogleCredentials.getApplicationDefault())
                .build()
                .getService();
        Blob blob = storage.get("your-bucket-name", "pth.to.your.kafka.client.truststore.jks");
        ReadChannel readChannel = blob.reader();
        FileOutputStream fileOuputStream;
        fileOuputStream = new FileOutputStream("/tmp/kafka.client.truststore.jks"); //path where the jks file will be stored
        fileOuputStream.getChannel().transferFrom(readChannel, 0, Long.MAX_VALUE);
        fileOuputStream.close();
        File f = new File("/tmp/kafka.client.truststore.jks"); //assuring the store file exists
        if (f.exists())
        {
            LOG.debug("key exists");
        }
        else
        {
            LOG.error("key does not exist");
        }
    } catch (FileNotFoundException e) {
        // TODO Auto-generated catch block
        LOG.error( e.getMessage());
    } catch (IOException e) {
        // TODO Auto-generated catch block
        LOG.error( e.getMessage());
    }
    config.put("ssl.truststore.location",(Object) "/tmp/kafka.client.truststore.jks" );
    return new KafkaConsumer<byte[], byte[]>(config);
}
}
</code>


  
    并且不要忘记在你的KafkaIO.read（）调用中使用.withConsumerFactoryFn，应该是这样的：
  

   
    
      Map configMap = new HashMap();
configMap.put(“security.protocol”, (Object) “SSL”);
configMap.put(“ssl.truststore.password”, (Object) “clientpass”);
p.apply(“ReadFromKafka”, KafkaIO.read()
            .withBootstrapServers(“ip:9093”)
            .withTopic(“pageviews”)
            .withKeyDeserializer(StringDeserializer.class)
            .withValueDeserializer(StringDeserializer.class)
            .updateConsumerProperties(configMap)
            .withConsumerFactoryFn(new ConsumerFactoryFn()) … etc.
</code>