CREATE APPLICATION KafkaWriterApplication;

CREATE SOURCE FileReader1 USING Global.FileReader ( 
  rolloverstyle: 'Default', 
  wildcard: 'posdata.csv', 
  blocksize: 64, 
  skipbom: true, 
  directory: '/Users/priyankasundararajan/Product/Samples/AppData', 
  includesubdirectories: false, 
  positionbyeof: false ) 
PARSE USING Global.DSVParser ( 
  trimwhitespace: false, 
  commentcharacter: '', 
  linenumber: '-1', 
  columndelimiter: ',', 
  trimquote: true, 
  columndelimittill: '-1', 
  ignoreemptycolumn: false, 
  separator: ':', 
  quoteset: '\"', 
  charset: 'UTF-8', 
  ignoremultiplerecordbegin: 'true', 
  ignorerowdelimiterinquote: false, 
  header: false, 
  blockascompleterecord: false, 
  rowdelimiter: '\n', 
  nocolumndelimiter: false, 
  headerlineno: 0 ) 
OUTPUT TO Stream1;

CREATE TARGET kafkawriter1 USING Global.KafkaWriter VERSION '2.1.0'( 
  brokerAddress: 'pkc-4yyd6.us-east1.gcp.confluent.cloud:9092', 
  KafkaConfigValueSeparator: '++', 
  MessageKey: '', 
  MessageHeader: '',  
  ParallelThreads: '', 
  Topic: 'kafkaDSVTest', 
  KafkaConfig: 'max.request.size++10485760|batch.size++10000120|security.protocol++SASL_SSL|ssl.endpoint.identification.algorithm++https|sasl.mechanism++PLAIN|sasl.jaas.config++org.apache.kafka.common.security.plain.PlainLoginModule required username=\"<CLUSTER_API_KEY>\" password=\"<CLUSTER_API_SECRET>\";', 
  KafkaConfigPropertySeparator: '|', 
  Mode: 'ASync' ) 
FORMAT USING Global.AvroFormatter  ( 
  schemaFileName: 'schema1', 
  formatAs: 'default' ) 
INPUT FROM Stream1;

END APPLICATION KafkaWriterApplication;

@Sid can we use Striim’s vault to store the cluster API key and password so it’s not in plain text of the kafkawriter target?

https://www.striim.com/docs/en/reading-from-and-writing-to-kafka-using-avro.html

You can store any adapter property value in a vault.

https://www.striim.com/docs/en/using-vaults.html