Package org.jumpmind.symmetric.io.data.writer

Class KafkaWriter

java.lang.Object

org.jumpmind.symmetric.io.data.writer.AbstractDatabaseWriter

org.jumpmind.symmetric.io.data.writer.DefaultDatabaseWriter

org.jumpmind.symmetric.io.data.writer.DynamicDefaultDatabaseWriter

org.jumpmind.symmetric.io.data.writer.KafkaWriter

All Implemented Interfaces:

IDataResource, IDataWriter

public class KafkaWriter
extends DynamicDefaultDatabaseWriter

Nested Class Summary

Nested classes/interfaces inherited from class org.jumpmind.symmetric.io.data.writer.AbstractDatabaseWriter

AbstractDatabaseWriter.LoadStatus

Field Summary

Fields

Modifier and Type	Field	Description
static final String	AVRO_CDC_SCHEMA
static final String	KAFKA_FORMAT_AVRO
static final String	KAFKA_FORMAT_CSV
static final String	KAFKA_FORMAT_JSON
static final String	KAFKA_FORMAT_XML
static final String	KAFKA_MESSAGE_BY_BATCH
static final String	KAFKA_MESSAGE_BY_ROW
static final String	KAFKA_SECURITY_PROTOCOL_PLAINTEXT
static final String	KAFKA_SECURITY_PROTOCOL_SASL_PLAINTEXT
static final String	KAFKA_SECURITY_PROTOCOL_SASL_SSL
static final String	KAFKA_SECURITY_PROTOCOL_SSL
protected final String	KAFKA_TEXT_CACHE
static final String	KAFKA_TOPIC_BY_CHANNEL
static final String	KAFKA_TOPIC_BY_TABLE
protected String	kafkaDataKey
protected Map<String,List<org.apache.kafka.clients.producer.ProducerRecord<String,Object>>>	kafkaDataMap
org.apache.kafka.clients.producer.KafkaProducer<String,Object>	kafkaProducer
protected static Map<String,org.apache.kafka.clients.producer.KafkaProducer<String,Object>>	producerMap

Fields inherited from class org.jumpmind.symmetric.io.data.writer.DynamicDefaultDatabaseWriter

tablePrefix, targetPlatform, targetTransaction

Fields inherited from class org.jumpmind.symmetric.io.data.writer.DefaultDatabaseWriter

CUR_DATA, currentDmlStatement, currentDmlValues, hasUncommittedDdl, isCteExpression, logSqlBuilder, platform, transaction

Fields inherited from class org.jumpmind.symmetric.io.data.writer.AbstractDatabaseWriter

batch, CONFLICT_ERROR, CONFLICT_IGNORE, conflictResolver, context, isRequiresSavePointsInTransaction, lastApplyChangesOnly, lastData, lastUseConflictDetection, missingTables, sourceTable, statistics, targetColumnSourceReferencesMap, targetTable, targetTables, TRANSACTION_ABORTED, uncommittedCount, writerSettings

Constructor Summary

Constructors

Constructor

Description

KafkaWriter(IDatabasePlatform symmetricPlatform, IDatabasePlatform targetPlatform, String prefix, IDatabaseWriterConflictResolver conflictResolver, DatabaseWriterSettings settings, String producer, String outputFormat, String topicBy, String messageBy, String confluentUrl, String schemaPackage, String externalNodeID, String url, String loadOnlyPrefix, TypedProperties props, String runtimeConfigTablePrefix, String channelReload)

Method Summary

Modifier and Type	Method	Description
protected void	allowInsertIntoAutoIncrementColumns(boolean value, Table table)
void	batchComplete(DataContext context)
protected boolean	create(CsvData data)
static byte[]	datumToByteArray(org.apache.avro.Schema schema, org.apache.avro.generic.GenericRecord datum)
protected AbstractDatabaseWriter.LoadStatus	delete(CsvData data, boolean useConflictDetection)
void	end(Batch batch, boolean inError)
protected int	execute(CsvData data, String[] values)
Class<?>	getClassByTableName(String tableName)
String	getColumnName(String dbTableName, String dbColumnName, Object bean)
String	getTableName(String dbTableName)
protected void	logFailureDetails(Throwable e, CsvData data, boolean logLastDmlDetails)
protected Table	lookupTableAtTarget(Table sourceTable)
protected void	prepare()
protected void	prepare(String sql, CsvData data)
int	prepareAndExecute(String sql, CsvData data)
void	sendKafkaMessage(org.apache.kafka.clients.producer.ProducerRecord<String,Object> record)
protected boolean	sql(CsvData data)
protected AbstractDatabaseWriter.LoadStatus	update(CsvData data, boolean applyChangesOnly, boolean useConflictDetection)
int	writeKafka(CsvData data, Table table)

Methods inherited from class org.jumpmind.symmetric.io.data.writer.DynamicDefaultDatabaseWriter

close, commit, getPlatform, getPlatform, getPlatform, getTablePrefix, getTargetPlatform, getTargetTransaction, getTransaction, getTransaction, getTransaction, isLoadOnly, isSymmetricTable, isSymmetricTable, open, rollback

Methods inherited from class org.jumpmind.symmetric.io.data.writer.DefaultDatabaseWriter

bindVariables, checkTargetTableHasColumns, commit, containsNullLookupKeyDataSinceLastStatement, create, dmlValuesToString, doesColumnNeedUpdated, end, getCurData, getCurrentDmlStatement, getSqlStatements, getTableKey, getWriterSettings, insert, isCteExpression, isDml, logDataTruncation, logFailureDetails, lookupTableFromCache, preprocessSqlStatement, putTableInCache, quoteString, removeGeneratedColumns, replaceCteExpression, replaceCteExpression, requireNewStatement, start, targetTableWasChangedByFilter, updateChangedDataIndicator

Methods inherited from class org.jumpmind.symmetric.io.data.writer.AbstractDatabaseWriter

buildTargetColumnReferencesMap, checkForEarlyCommit, clearTargetColumnReferencesMap, filterAfter, filterBefore, filterError, getBatch, getConflictResolver, getContext, getLookupDataMap, getPkDataFor, getRowData, getSourceTable, getStatistics, getTargetColumnReferencesMap, getTargetTable, hasFilterThatHandlesMissingTable, notifyFiltersBatchCommitted, notifyFiltersBatchComplete, notifyFiltersBatchRolledback, notifyFiltersEarlyCommit, refreshTargetColumnReferencesMap, script, setConflictResolver, start, write

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

KAFKA_TEXT_CACHE

protected final String KAFKA_TEXT_CACHE

kafkaDataMap

protected Map<String,List<org.apache.kafka.clients.producer.ProducerRecord<String,Object>>> kafkaDataMap

kafkaDataKey

protected String kafkaDataKey

KAFKA_FORMAT_XML

public static final String KAFKA_FORMAT_XML

See Also:

• Constant Field Values

KAFKA_FORMAT_JSON

public static final String KAFKA_FORMAT_JSON

See Also:

• Constant Field Values

KAFKA_FORMAT_AVRO

public static final String KAFKA_FORMAT_AVRO

See Also:

• Constant Field Values

KAFKA_FORMAT_CSV

public static final String KAFKA_FORMAT_CSV

See Also:

• Constant Field Values

KAFKA_MESSAGE_BY_BATCH

public static final String KAFKA_MESSAGE_BY_BATCH

See Also:

• Constant Field Values

KAFKA_MESSAGE_BY_ROW

public static final String KAFKA_MESSAGE_BY_ROW

See Also:

• Constant Field Values

KAFKA_TOPIC_BY_TABLE

public static final String KAFKA_TOPIC_BY_TABLE

See Also:

• Constant Field Values

KAFKA_TOPIC_BY_CHANNEL

public static final String KAFKA_TOPIC_BY_CHANNEL

See Also:

• Constant Field Values

AVRO_CDC_SCHEMA

public static final String AVRO_CDC_SCHEMA

See Also:

• Constant Field Values

KAFKA_SECURITY_PROTOCOL_PLAINTEXT

public static final String KAFKA_SECURITY_PROTOCOL_PLAINTEXT

See Also:

• Constant Field Values

KAFKA_SECURITY_PROTOCOL_SASL_PLAINTEXT

public static final String KAFKA_SECURITY_PROTOCOL_SASL_PLAINTEXT

See Also:

• Constant Field Values

KAFKA_SECURITY_PROTOCOL_SASL_SSL

public static final String KAFKA_SECURITY_PROTOCOL_SASL_SSL

See Also:

• Constant Field Values

KAFKA_SECURITY_PROTOCOL_SSL

public static final String KAFKA_SECURITY_PROTOCOL_SSL

See Also:

• Constant Field Values

kafkaProducer

public org.apache.kafka.clients.producer.KafkaProducer<String,Object> kafkaProducer

producerMap

protected static Map<String,org.apache.kafka.clients.producer.KafkaProducer<String,Object>> producerMap

Constructor Details

KafkaWriter

public KafkaWriter(IDatabasePlatform symmetricPlatform,
 IDatabasePlatform targetPlatform,
 String prefix,
 IDatabaseWriterConflictResolver conflictResolver,
 DatabaseWriterSettings settings,
 String producer,
 String outputFormat,
 String topicBy,
 String messageBy,
 String confluentUrl,
 String schemaPackage,
 String externalNodeID,
 String url,
 String loadOnlyPrefix,
 TypedProperties props,
 String runtimeConfigTablePrefix,
 String channelReload)

Method Details

prepare

protected void prepare()

Overrides:

prepare in class DefaultDatabaseWriter

prepare

protected void prepare(String sql,
 CsvData data)

Overrides:

prepare in class DefaultDatabaseWriter

prepareAndExecute

public int prepareAndExecute(String sql,
 CsvData data)

Overrides:

prepareAndExecute in class DefaultDatabaseWriter

execute

protected int execute(CsvData data,
 String[] values)

Overrides:

execute in class DefaultDatabaseWriter

delete

protected AbstractDatabaseWriter.LoadStatus delete(CsvData data,
 boolean useConflictDetection)

Overrides:

delete in class DefaultDatabaseWriter

update

protected AbstractDatabaseWriter.LoadStatus update(CsvData data,
 boolean applyChangesOnly,
 boolean useConflictDetection)

Overrides:

update in class DefaultDatabaseWriter

create

protected boolean create(CsvData data)

Overrides:

create in class DefaultDatabaseWriter

sql

protected boolean sql(CsvData data)

Overrides:

sql in class DefaultDatabaseWriter

logFailureDetails

protected void logFailureDetails(Throwable e,
 CsvData data,
 boolean logLastDmlDetails)

Overrides:

logFailureDetails in class DefaultDatabaseWriter

allowInsertIntoAutoIncrementColumns

protected void allowInsertIntoAutoIncrementColumns(boolean value,
 Table table)

Overrides:

allowInsertIntoAutoIncrementColumns in class DefaultDatabaseWriter

lookupTableAtTarget

protected Table lookupTableAtTarget(Table sourceTable)

Overrides:

lookupTableAtTarget in class DefaultDatabaseWriter

end

public void end(Batch batch,
 boolean inError)

Specified by:

end in interface IDataWriter

Overrides:

end in class DefaultDatabaseWriter

getTableName

public String getTableName(String dbTableName)

getClassByTableName

public Class<?> getClassByTableName(String tableName)

getColumnName

public String getColumnName(String dbTableName,
 String dbColumnName,
 Object bean)

sendKafkaMessage

public void sendKafkaMessage(org.apache.kafka.clients.producer.ProducerRecord<String,Object> record)

datumToByteArray

public static byte[] datumToByteArray(org.apache.avro.Schema schema,
 org.apache.avro.generic.GenericRecord datum)
                               throws IOException

Throws:

IOException

batchComplete

public void batchComplete(DataContext context)

writeKafka

public int writeKafka(CsvData data,
 Table table)