Delivery Guarantees

The AMPS Kafka sink publishes Kafka record values to AMPS. The main delivery guarantees are provided by the AMPS publish store configuration.

Publish Stores

The sink can create an AMPS publish store through pubStoreType.

Store Type	Description
memory	Stores outgoing publishes in the worker process. This is fast, but does not protect against process failure.
file	Stores outgoing publishes on disk. This can recover after client or worker failure and depends on the performance of the storage device.

The following configuration uses a memory publish store:

{
    "name": "amps-kafka-sink",
    "config": {
        "connector.class": "com.crankuptheamps.kafka.AMPSKafkaSink",
        "topics": "Orders",
        "tasks.max": "1",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
        "header.converter": "org.apache.kafka.connect.storage.StringConverter",
        "clientFactoryClass": "com.crankuptheamps.kafka.AMPSBasicClientFunction",
        "clientName": "KafkaSink",
        "pubStoreType": "memory",
        "pubStoreInitialCap": "100",
        "uri": "tcp://localhost:9007/amps/json",
        "ampsTopic": "AMPSKafkaSinkTest",
        "maxBatch": "100",
        "useTopicHeader": "false",
        "publishFlushTimeout": "0"
    }
}

The following properties use a file-backed publish store:

{
    "pubStoreType": "file",
    "pubStoreInitialCap": "1000",
    "pubStorePath": "./KafkaSink.PubStore",
    "pruneTimeThreshold": "300000"
}

Failed Writes

The sink registers an AMPS failed write handler. When AMPS reports a failed write, the connector logs the reason. If logMsgOnWriteError is true, the connector also logs the message that failed when the AMPS client has enough stored information to provide it.