Skip to main content

Kafka Writer Programmer's Reference

Kafka Writer properties

The following table lists the Kafka Writer adapter properties.

property

type

default value

notes

Auto Create Topic

Boolean

True

With the default value of True, when a topic with the expected name does not exist, Kafka Writer will create it automatically.

  • If Topics is a wildcard, the names will be defined by the Topic Key.

  • If Topics is a user-defined list, the names will be taken from the list.

Checkpoint Topic

String

Appears in UI when Exactly Once Processing is True.

Name of the checkpointing topic. By default the name of the checkpoint topic will be <writer name>_CHECKPOINT. You may edit this property to change the name of the checkpointing topic. The checkpoint topic must be unique for each writer.

It is mandatory to have log.cleanup.policy set to compact, the retention time set high enough to accommodate the maximum downtime of the Striim application, and replication factor greater than 1.

If the checkpoint topic does not exist, it will be created automatically. Striim must have the necessary permissions to create and delete the checkpoint topic.

Checkpoint Topic Config

String

"{"PartitionCount":1, "ReplicationFactor": 3, "CleanUpPolicy":"compact", "min.cleanable.dirty.ratio":"0.5", "segment.ms":"86400000", "segment.bytes":"1073741824", "min.compaction.lag.ms":"3600000", "max.compaction.lag.ms":"604800000"}"

Appears in UI when Exactly Once Processing is True.

By default the checkpoint topic will be created with a single partition, replication factor of 3, and log cleanup policy of compact. You may override the settings if required with the following considerations:

  • Replication factor: set to a higher value for higher availability.

  • Cleanup policy: do not modify unless instructed to by Striim Support.

  • If Parallel Threads exceeds the PartitionCount value, the partition count will be automatically adjusted to match the Parallel Threads value.

  • If Partition Count and Replication Factor are not specified, the values for "num.partitions" and "default.replication.factor" from the Kafka broker configuration will be used. If these values are not available in the broker configuration, default fallback values will be applied.

Commit Policy

String

EventCount=10000, Interval=15s

The commit policy controls how often transactions are committed in the target. If only EventCount is specified, a batch will be sent when specified number is reached. If only Interval is specified, a batch will be sent every time the interval (default 15 seconds) has elapsed. If both are sepcified, a batcn will be sent when either limit is reached.

A batch will always be sent whenever a DDL event is received.

Connection Profile Name

Enum

Specify the name of the Kafka connection profile to use, or select New Connection Profile to create one. For mroe information, see Connection profiles.Connection profiles

Connection Retry

String

initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=5, totalTimeout=10m

With the default setting, if a connection attempt is unsuccessful, the adapter will try again in 10 seconds (InitialRetryDelay=10s). If the second attempt is unsuccessful, in 20 seconds it will try a third time (InitialRetryDelay=10s multiplied by retryDelayMultiplier=2). If that fails, the adapter will try again in 40 seconds (the previous retry interval 20s multiplied by 2). If connection attempts continue to fail, the the adapter will try again every 60 seconds (maxRetryDelay=1m) until a total of five connection attempts have been made (maxAttempts=5), after which the adapter will halt and log an exception.

The adapter will halt when either maxAttempts or totalTimeout is reached.

InitialRetryDelay, maxRetryDelay, and totalTimeout may be specified in milliseconds (ms), seconds (s, the default), or minutes (m).

If retryDelayMultiplier is set to 1, connection will be attempted on the fixed interval set by InitialRetryDelay.

To disable connection retry, set maxAttempts=0.

Negative values are not supported.

Custom Message Key

Property-value pairs

Appears in UI when Message Key is Custom.

Specify one or more key-value pairsto be sent as the Kafka message key. The value can be a static string or a field from the incoming stream. The format of the property must be Key=Value; Key=value, for example, OperationCode=@metadata(OperationName); SourceTableName=@metadata(TableName).

Custom Partition Key

String

Appears in UI when Partition Key is Custom.

Specify the name of a field from an input stream of a user-defined type, or a metadata or userdata field in an input sttream of type AVROEvent, JSONNodeEvent, or WAEvent.

Data Topic Config

When Auto Create Topic is True, optionally specify Kafka properties to be used when creating topics, for example, '{"PartitionCount":1, "ReplicationFactor":3, "CleanUp Policy":"compact", "Retention Time":"2592000000", "Retention Size":"1048576"}'

Exactly Once Processing

Boolean

True

See E1P (exactly-once processing) and A1P (at-least-once processing).E1P (exactly-once processing)A1P (at-least-once processing)

Formatter

Enum

AvroFormatter

This adapter has a choice of formatters. See Formatters for more information. For information on the formatter's advanced properties shown in the Flow Designer, click View Documentation to see the documentation for the selected formatter.

Message Header

String

A property that defines how Kafka message headers are constructed for each event. If left blank, no additional headers are added. Optionally, specify one or more key-value pairs, where each value can be a static string or a field from the input stream, for example, OperationName=@metadata(OperationName); CompanyName=@userdata(companyName). For more information, see Message Header use cases and examples.Message Header use cases and examples

Message Key

String

None

With the default value of None, no message key will be sent.

Set to Custom to create the message keys based on one or more fields from the input stream.

Set to Primary Key to create the message keys based on the values of the incoming event's primary key fields. This is applicable only for compatible database and data warehouse sources.

For more information, see Message Key use cases and examples.Message Key use cases and examples

Parallel Threads

Integer

See Creating multiple writer instances (parallel threads).

Partition Key

Enum

None

With the default value of None, no explicit partition key is set.

Set to Custom to define a partition key based on a field from the input stream.

Set to Message Key to define partition keys based on the message keys.

For more information, see Partition Key use cases and examples.Partition Key use cases and examples

Persist Schema

Enum

ON

Appears in UI when Schema Evolution is Auto.

With the default value of ON, DDL events will be will be written to the target as Kafka messages. Set to OFF if you do not want those messages to be written to the target.

In 5.4.0.2 and later releases, set to MSGHDR to include the SQL parsed data in the header of the first DML event after a DDL event. This option is usable only when the formatter is Avro Parser and recovery is enabled. With this option:

  • The header contains DDL schema information only in the first DML event after a DDL event.

  • Subsequent DML events do not carry the DDL schema in their headers, except that after a restart, if processing begins with the first DML event after a DDL, that event's header will carry the DDL schema.

  • If the schema is registered externally, the DDL information will not be maintained in the upcoming DML events' headers.

  • If multiple DDL events precede a DML event, the DML header includes all preceding DDLs. However, on restart, only the DDL immediately before the DML is preserved in that event's header.

Schema Evolution

Enum

Auto

With the default value of Auto, the associated schema registry will be evolved with supported DDL events. (This is equivalent to CDDL Action = Process in other adapters.) Set to Manual to halt the application when a DDL event is encountered. (This is equivalent to CDDL Action = Halt in other adapters.) For more information, see Handling schema evolution.

Schema File Name

String

See Avro Formatter and Schema registry use cases and examples.

Serializer

Enum

Striim Serializer

If the Formatter is AvroFormatter and the downstream application expects Kafka messages written in Confluent wire format, select Confluent Serializer. Otherwise, leave set to default. For more information, see Message formats.Message formats

Topic Key for SmartMap (TopicKey in TQL)

If a wildcard or more than one topic is specified in the Topics property, specify the name of a field in the input stream to be used to distribute events among topics. For more information, see Topics and Topic Key use cases and examples.Topics and Topic Key use cases and examples

Topic Names for SmartMap (Topics in TQL)

String

Defines the names of the topic(s) to be written to. For more information, see Topics and Topic Key use cases and examples.Topics and Topic Key use cases and examples

Use Schema Registry Connection Profile

Boolean

False

Set to True to use a connection profile for the schema registry connection properties.

Kafka Connection Profile Properties

The following table lists the properties for the Kafka Connection Profile.

Property Name

Required

Default

Alterable

Description

BrokerAddress

TRUE

FALSE

Comma-separated list of host:port pairs for establishing the initial connection to the Kafka cluster (bootstrap.servers). All addresses must belong to the same cluster.

AuthenticationType

TRUE

NONE

FALSE

Authentication type. Options: AWS_MSK_IAM, NONE, Mutual TLS, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI.

UseSSL

FALSE

True

FALSE

Toggle to enable SSL for server authentication. Set to True when connecting to Confluent Cloud.

UseCertificate

FALSE

FALSE

FALSE

Toggle to enable PEM-based TrustStore and KeyStore files instead of JKS keystore/truststore.

CACertificate

FALSE (required when UseSSL=True and UseCertificate=True)

TRUE

File path or Vault reference for the CA (Certificate Authority) certificate. Used to validate the server's identity. Maps to ssl.truststore.certificates.

SSLTruststoreLocation

FALSE (required when UseSSL=True and UseCertificate=False)

TRUE

File path to the truststore file containing trusted CA certificates. Maps to ssl.truststore.location.

SSLTruststorePassword

FALSE (required when UseSSL=True and UseCertificate=False)

TRUE

Password for the Truststore file. Maps to ssl.truststore.password.

SSLKeyStoreLocation

FALSE

TRUE

Path to the keystore file containing the client's certificate and private key. Maps to ssl.keystore.location.

SSLKeyStorePassword

FALSE (required when AuthType=Mutual TLS and UseCertificate=False)

TRUE

Password for the Keystore file. Maps to ssl.keystore.password.

SSLKeystoreCertificateChain

FALSE (required when AuthType=Mutual TLS and UseCertificate=True)

TRUE

PEM-formatted client certificate chain for mTLS authentication. Maps to ssl.keystore.certificate.chain.

SSLKeyStoreKey

FALSE (required when AuthType=Mutual TLS and UseCertificate=True)

TRUE

PEM-formatted private key for mTLS authentication. Maps to ssl.keystore.key.

SSLKeyPassword

FALSE (required when AuthType=Mutual TLS)

TRUE

Password to decrypt the private key inside the keystore. Maps to ssl.key.password.

JaasConfig

FALSE (required for PLAIN, SCRAM-*, GSSAPI)

TRUE

JAAS configuration string for SASL authentication (sasl.jaas.config). Supports Vault references.

AdditionalConnectionConfig

FALSE

TRUE

Additional Kafka connection configurations as JSON key-value pairs. Example: '{"session.timeout.ms":"30000"}'.

ProducerConfig

FALSE

TRUE

Additional Kafka producer configurations as JSON. Default values set: enable.idempotence=true, acks=all, retries, max.in.flight.requests.per.connection=5.

ConsumerConfig

FALSE

TRUE

Additional Kafka consumer configurations as JSON. Defaults: max.partition.fetch.bytes=10485760, fetch.min.bytes=1048576, fetch.max.wait.ms=1000.

Data Type Mapping

The following sections describe how source database column types are mapped to Avro data types when using the Avro Formatter.

When the Schema Conversion Tool (SCT) is used and the source schema flows through the pipeline, closer logical Avro type mapping is generated. For example, a numeric column with precision and scale information produces:

{"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 0}

For Time/Timestamp types using Java 8-based data types (up to nanosecond precision; Avro supports up to microseconds):

  • Precision ≤ 6: mapped to time-micros, timestamp-micros, or local-timestamp-micros.

  • Precision > 6: falls back to String.

Oracle to Avro Data Type Mapping

Source Type

Java Type (TypeUUID)

Native (UUID)

Native (SCT)

Logical (UUID)

Logical (SCT)

NUMBER(10)

java.lang.String

string

bytes

string

decimal(p=10, s=0)

INTEGER

java.lang.String

string

bytes

string

decimal(p=38, s=0)

NUMBER(38,19)

java.lang.String

string

bytes

string

decimal(p=38, s=19)

NUMBER(10,2)

java.lang.String

string

bytes

string

decimal(p=10, s=2)

NUMBER

java.lang.String

string

string

string

string

BINARY_FLOAT

java.lang.Float

float

float

float

float

BINARY_DOUBLE

java.lang.Double

double

bytes

double

bytes

FLOAT(32)

java.lang.String

string

double

string

double

CHAR(2000)

java.lang.String

string

string

string

string

VARCHAR2(4000)

java.lang.String

string

string

string

string

NCHAR(1000)

java.lang.String

string

string

string

string

NVARCHAR2(2000)

java.lang.String

string

string

string

string

DATE

org.joda.time.DateTime

string

int

string

date

TIMESTAMP(0)

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(3)

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(6)

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(7)

org.joda.time.DateTime

string

string

string

string

TIMESTAMP(9)

org.joda.time.DateTime

string

string

string

string

TIMESTAMP(0) WITH TIME ZONE

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(6) WITH TIME ZONE

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(7) WITH TIME ZONE

org.joda.time.DateTime

string

string

string

string

TIMESTAMP WITH TIME ZONE

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(0) WITH LOCAL TIME ZONE

org.joda.time.DateTime

string

long

string

local-timestamp-micros

TIMESTAMP(6) WITH LOCAL TIME ZONE

org.joda.time.DateTime

string

long

string

local-timestamp-micros

TIMESTAMP(7) WITH LOCAL TIME ZONE

org.joda.time.DateTime

string

string

string

string

TIMESTAMP WITH LOCAL TIME ZONE

org.joda.time.DateTime

string

long

string

local-timestamp-micros

INTERVAL YEAR TO MONTH

java.lang.String

string

fixed

string

duration

INTERVAL DAY TO SECOND

java.lang.String

string

fixed

string

duration

RAW(2000)

java.lang.String

string

bytes

string

bytes

ROWID

java.lang.String

string

string

string

string

CLOB

java.lang.String

string

string

string

string

NCLOB

java.lang.String

string

string

string

string

BLOB

java.lang.String

string

bytes

string

bytes

MySQL to Avro Data Type Mapping

Source Type

Java Type (TypeUUID)

Native (UUID)

Native (SCT)

Logical (UUID)

Logical (SCT)

INTEGER

java.lang.Integer

int

int

int

int

TINYINT

java.lang.Byte

int

int

int

int

SMALLINT

java.lang.Short

int

int

int

int

MEDIUMINT

java.lang.Integer

int

int

int

int

BIGINT

java.lang.Long

long

int

long

int

FLOAT

java.lang.Float

float

float

float

float

DOUBLE

java.lang.Double

double

double

double

double

DECIMAL(65,30)

java.lang.String

string

string

string

string

NUMERIC(10,2)

java.lang.String

string

bytes

string

decimal(p=10, s=2)

CHAR(255)

java.lang.String

string

string

string

string

VARCHAR(1000)

java.lang.String

string

string

string

string

BINARY(255)

byte[]

string

bytes

bytes

bytes

VARBINARY(16383)

byte[]

string

bytes

bytes

bytes

TINYTEXT

java.lang.String

string

string

string

string

TEXT

java.lang.String

string

string

string

string

MEDIUMTEXT

java.lang.String

string

string

string

string

LONGTEXT

java.lang.String

string

string

string

string

TINYBLOB

byte[]

string

bytes

bytes

bytes

BLOB

byte[]

string

bytes

bytes

bytes

MEDIUMBLOB

byte[]

string

bytes

bytes

bytes

LONGBLOB

byte[]

string

bytes

bytes

bytes

DATE

org.joda.time.LocalDate

string

int

string

date

DATETIME(0)

org.joda.time.DateTime

string

long

string

timestamp-micros

DATETIME(3)

org.joda.time.DateTime

string

long

string

timestamp-micros

DATETIME(6)

org.joda.time.DateTime

string

long

string

timestamp-micros

DATETIME

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(0)

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(3)

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(6)

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP

org.joda.time.DateTime

string

string

string

string

TIME(0)

java.lang.String

string

int

string

time-micros

TIME(3)

java.lang.String

string

int

string

time-micros

TIME(6)

java.lang.String

string

int

string

time-micros

TIME

java.lang.String

string

string

string

string

YEAR

java.lang.Short

int

int

int

int

ENUM(...)

java.lang.Integer

int

string

int

string

SET(...)

java.lang.Long

long

string

long

string

BOOLEAN

java.lang.Boolean

boolean

int

boolean

int

JSON

java.lang.String

string

string

string

string

BIT(64)

java.lang.String

string

bytes

string

bytes

GEOMETRY

java.lang.String

string

string

string

string

PostgreSQL to Avro Data Type Mapping

Source Type

Java Type (TypeUUID)

Native (UUID)

Native (SCT)

Logical (UUID)

Logical (SCT)

BIGSERIAL

java.lang.Long

long

int

long

int

SMALLINT

java.lang.Short

int

int

int

int

INTEGER

java.lang.Integer

int

int

int

int

BIGINT

java.lang.Long

long

int

long

int

SERIAL

java.lang.Integer

int

int

int

int

DECIMAL(18,6)

java.lang.String

string

bytes

string

decimal(p=18, s=6)

NUMERIC(15,5)

java.lang.String

string

bytes

string

decimal(p=15, s=5)

REAL

java.lang.Float

float

float

float

float

DOUBLE PRECISION

java.lang.Double

double

double

double

double

CHAR(100)

java.lang.String

string

string

string

string

VARCHAR(1000)

java.lang.String

string

string

string

string

TEXT

java.lang.String

string

string

string

string

BOOLEAN

java.lang.Short

int

boolean

int

boolean

DATE

org.joda.time.LocalDate

string

int

string

date

TIME(0)

java.lang.String

string

int

string

time-micros

TIME(3)

java.lang.String

string

int

string

time-micros

TIME(6)

java.lang.String

string

int

string

time-micros

TIME

java.lang.String

string

string

string

string

timetz(0)

java.lang.String

string

string

string

string

timetz(6)

java.lang.String

string

string

string

string

TIMESTAMP(0)

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(3)

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP(6)

org.joda.time.DateTime

string

long

string

timestamp-micros

TIMESTAMP

org.joda.time.DateTime

string

long

string

timestamp-micros

timestamptz(0)

org.joda.time.DateTime

string

long

string

timestamp-micros

timestamptz(6)

org.joda.time.DateTime

string

long

string

timestamp-micros

timestamptz

org.joda.time.DateTime

string

long

string

timestamp-micros

INTERVAL

java.lang.String

string

fixed

string

duration

BYTEA

java.lang.String

string

string

string

string

INET

java.lang.String

string

string

string

string

CIDR

java.lang.String

string

string

string

string

UUID

java.lang.String

string

string

string

uuid

JSON

java.lang.String

string

string

string

string

JSONB

java.lang.String

string

string

string

string

XML

java.lang.String

string

string

string

string

MONEY

java.lang.String

string

string

string

string

BIT(8)

java.lang.String

string

bytes

string

bytes

VARBIT(100)

java.lang.String

string

bytes

string

bytes

SQL Server to Avro Data Type Mapping

Source Type

Java Type (TypeUUID)

Native (UUID)

Native (SCT)

Logical (UUID)

Logical (SCT)

INT IDENTITY(1,1)

java.lang.Integer

int

int

int

int

TINYINT

java.lang.Short

int

int

int

int

SMALLINT

java.lang.Short

int

int

int

int

INT

java.lang.Integer

int

int

int

int

BIGINT

java.lang.Long

long

int

long

int

DECIMAL(38,19)

java.lang.String

string

bytes

string

decimal(p=38, s=19)

NUMERIC(18,6)

java.lang.String

string

bytes

string

decimal(p=18, s=6)

MONEY

java.lang.String

string

string

string

string

SMALLMONEY

java.lang.String

string

string

string

string

FLOAT(24)

java.lang.Double

double

double

double

double

REAL

java.lang.Float

float

float

float

float

CHAR(100)

java.lang.String

string

string

string

string

VARCHAR(1000)

java.lang.String

string

string

string

string

NCHAR(100)

java.lang.String

string

string

string

string

NVARCHAR(1000)

java.lang.String

string

string

string

string

TEXT

java.lang.String

string

string

string

string

NTEXT

java.lang.String

string

string

string

string

DATE

org.joda.time.LocalDate

string

int

string

date

TIME(0)

java.lang.String

string

int

string

time-micros

TIME(3)

java.lang.String

string

int

string

time-micros

TIME(6)

java.lang.String

string

int

string

time-micros

TIME(7)

java.lang.String

string

string

string

string

DATETIME

org.joda.time.DateTime

string

long

string

timestamp-micros

DATETIME2(0)

org.joda.time.DateTime

string

long

string

timestamp-micros

DATETIME2(3)

org.joda.time.DateTime

string

long

string

timestamp-micros

DATETIME2(6)

org.joda.time.DateTime

string

long

string

timestamp-micros

DATETIME2(7)

org.joda.time.DateTime

string

string

string

string

DATETIME2

org.joda.time.DateTime

string

long

string

timestamp-micros

SMALLDATETIME

org.joda.time.DateTime

string

string

string

string

DATETIMEOFFSET(0)

java.lang.String

string

string

string

string

DATETIMEOFFSET(7)

java.lang.String

string

string

string

string

BINARY(100)

byte[]

string

bytes

bytes

bytes

VARBINARY(1000)

byte[]

string

bytes

bytes

bytes

BIT

java.lang.String

string

boolean

string

boolean

UNIQUEIDENTIFIER

java.lang.String

string

string

string

uuid

HIERARCHYID

java.lang.String

string

string

string

string

SQL_VARIANT

java.lang.String

string

string

string

string

XML

java.lang.String

string

string

string

string

GEOMETRY

java.lang.String

string

string

string

string

GEOGRAPHY

java.lang.String

string

string

string

string

Sample TQLs for Various Scenarios

Sample Source Schemas

The following schemas are referenced throughout TQL examples. If you are upgrading an existing application, also see TQL Examples Before and After Upgrade for a migration-focused comparison.

Sch.EMP

CREATE TYPE EmployeeType (
empId Integer KEY,
firstName String,
lastName String,
email String,
phoneNumber String,
jobId String,
managerId Integer,
deptId Integer
);

Sch.DEPT

CREATE TYPE DepartmentType (
deptId Integer KEY,
deptName String,
managerId Integer
);

Sales.Transaction

CREATE TYPE SalesTransactionType (
transactionId String KEY,
productId String,
customerId String,
saleAmount Double,
saleDate DateTime,
saleTime DateTime,
city String,
state String,
country String
);

Wildcard Topic Mapping (Auto Topic Creation, Confluent Serializer, E1P)

This example builds on the concepts introduced in Wildcard Topic Mapping. Auto topic creation, Primary Key as Message Key, Message Key as Partition Key, custom Message Header, wildcard multi-topic mapping, Confluent Serializer:

CREATE OR REPLACE APPLICATION DBToKafka;
CREATE OR REPLACE SOURCE DB_Source USING Global.MysqlReader (
DatabaseProviderType: 'Default',
FetchSize: 100,
adapterName: 'MysqlReader',
QuiesceOnILCompletion: false,
ConnectionURL: 'jdbc:mysql://localhost:3306/waction',
Tables: 'waction.%;',
Password: 'w@ct10n',
Username: 'root' )
OUTPUT TO DB_Data;
CREATE OR REPLACE TARGET KTWWildcardTopic USING Global.KafkaWriter (
CheckpointTopicConfig: '{"PartitionCount":1,"ReplicationFactor":3}',
adapterName: 'KafkaMultiTopicWriter',
Serializer: 'ConfluentSerializer',
PartitionKey: UseMessageKey,
MessageHeader: 'TableName=@MetaData(TableName)',
TopicKey: '@MetaData(TableName)',
AutoCreateTopic: true,
CDDLAction: 'Process',
E1P: true,
Topics: '%,%',
ConnectionProfileName: 'admin.KafkaCp',
CommitPolicy: 'EventCount=10000;Interval=15s',
MessageKey: 'PrimaryKey' )
FORMAT USING Global.AvroFormatter (
SchemaRegistryConnectionProfileName: 'admin.confluentSchemaRegistryCP',
useSchemaRegistryConnectionProfile: 'true',
formatAs: 'Table' )
INPUT FROM DB_Data;
END APPLICATION DBToKafka;

Explicit Topic Mapping, Multi-Partition

CREATE OR REPLACE APPLICATION DBToKafka;

CREATE OR REPLACE SOURCE DB_Source USING Global.DatabaseReader (
  Tables: 'waction.Bank;waction.Person;waction.retailData;',
  DatabaseProviderType: 'Default',
  FetchSize: 100,
  ConnectionURL: 'jdbc:mysql://localhost:3306/waction',
  Username: 'root',
  Password: '<password>' )
OUTPUT TO DB_Data;

CREATE CQ UserdataAdder
INSERT INTO Modified_DB_Data
SELECT putUserData(x,'JOB',data[2],'MARITAL',data[3]) FROM DB_Data x;;

CREATE OR REPLACE TARGET KTWMultiPartitionTopic USING Global.KafkaWriter (
  ConnectionProfileName: 'admin.KafkaCp',
  E1P: false,
  Serializer: 'ConfluentSerializer',
  MessageHeader: 'TableName=@MetaData(TableName)',
  Topics: 'waction.Bank,Bank;waction.Person,Person;waction.retailData,retailData',
  TopicKey: '@MetaData(TableName)',
  AutoCreateTopic: true,
  SchemaEvolution: 'Auto',
  DataTopicConfig: '{"PartitionCount":"3","ReplicationFactor":"4"}',
  PartitionKey: Custom,
  CustomPartitionKey: '@Metadata(TableName)',
  MessageKey: 'PrimaryKey',
  CommitPolicy: 'EventCount=10000;Interval=15s',
  CheckpointTopicConfig: '{"PartitionCount":1,"ReplicationFactor":3}' )
FORMAT USING Global.DSVFormatter (
  quotecharacter: '\"',
  columndelimiter: ',',
  nullvalue: 'NULL',
  usequotes: 'false',
  rowdelimiter: '\n',
  standard: 'none',
  header: 'false' )
INPUT FROM Modified_DB_Data;

END APPLICATION DBToKafka;

Single Topic Mapping

CREATE OR REPLACE APPLICATION FileToKafka;

CREATE OR REPLACE SOURCE FileSource USING Global.FileReader (
  directory: '/data/CSVData',
  wildcard: 'bank.csv',
  includesubdirectories: false )
PARSE USING Global.DSVParser (
  columndelimiter: ',',
  trimquote: true,
  quoteset: '\"',
  charset: 'UTF-8',
  rowdelimiter: '\n',
  header: true )
OUTPUT TO FileData;

CREATE OR REPLACE TARGET KTWSingleTopic USING Global.KafkaWriter (
  ConnectionProfileName: 'admin.KafkaCp',
  Topics: 'Employee',
  MessageKey: 'None',
  AutoCreateTopic: true,
  SchemaEvolution: 'Auto',
  E1P: true,
  Serializer: 'StriimSerializer',
  CommitPolicy: 'EventCount=10000;Interval=15s',
  CheckpointTopicConfig: '{"PartitionCount":1,"ReplicationFactor":3}' )
FORMAT USING Global.DSVFormatter (
  quotecharacter: '\"',
  columndelimiter: ',',
  nullvalue: 'NULL',
  usequotes: 'false',
  rowdelimiter: '\n',
  standard: 'none',
  header: 'false' )
INPUT FROM FileData;

END APPLICATION FileToKafka;

Explicit Topic Mapping with Avro Native Format

CREATE OR REPLACE APPLICATION DBToKafka;

CREATE OR REPLACE SOURCE DB_Source USING Global.DatabaseReader (
  Tables: 'waction.Bank;waction.Person;waction.retailData;',
  DatabaseProviderType: 'Default',
  FetchSize: 100,
  ConnectionURL: 'jdbc:mysql://localhost:3306/waction',
  Username: 'root',
  Password: '<password>' )
OUTPUT TO DB_Data;

CREATE OR REPLACE TARGET KTWExplicitTopic USING Global.KafkaWriter (
  ConnectionProfileName: 'admin.KafkaCp',
  MessageKey: 'Custom',
  CustomMessageKey: 'OperationName=@MetaData(OperationName)',
  TopicKey: '@MetaData(TableName)',
  Topics: 'waction.Bank,Bank;waction.Person,Person;waction.retailData,retailData',
  AutoCreateTopic: true,
  SchemaEvolution: 'Auto',
  DataTopicConfig: '{"PartitionCount":"1","ReplicationFactor":"3"}',
  PartitionKey: 'None',
  MessageHeader: 'CompanyName="Striim"',
  E1P: true,
  Serializer: 'StriimSerializer',
  CommitPolicy: 'EventCount=10000;Interval=15s',
  CheckpointTopicConfig: '{"PartitionCount":1,"ReplicationFactor":3}' )
FORMAT USING Global.AvroFormatter (
  useSchemaRegistryConnectionProfile: 'true',
  SchemaRegistryConnectionProfileName: 'admin.confluentSchemaRegistryCP',
  formatAs: 'Native',
  SchemaRegistrySubjectName: 'UseTopicName' )
INPUT FROM DB_Data;

END APPLICATION DBToKafka;