Send events to error handlers when broker is unavailable

[dilini-muthumala]

Purpose

Resolves #143

This is a new improvement.
Currently when the broker is not available, kafka producer will retry sending the event, but if the broker does not become available during that retry period, the event will be dropped.
After this improvement, such events can be redirected to Siddhi Error Handlers (STREAM/STORE/WAIT).

Goals

Add Error handling support for Kafka Sink

Approach

Throw ConnectionUnavailableException when TimeoutException occurrs.

Sample Configuration

Please refer sample siddhi app below. Here, when the broker is not available, we have specified to redirect the events to Error Store (on.error='STORE' )

@App:name("HTTPToKafka")
@App:description('Consume events from HTTP and publish to Kafka.')

@Source(type = 'http', receiver.url='http://localhost:8006/SweetProductionStream', basic.auth.enabled='false',
        @map(type='json'))
define stream SweetProductionStream (name string, amount double);

@sink(type='kafka',
          on.error='STORE',
          topic='bulk-orders',
          bootstrap.servers='localhost:9092',
--       partition.no='0',
          is.synchronous='true',
          retry.send='true',
          @map(type='json'))
    define stream BulkOrdersStream (name string, amount double);

from SweetProductionStream
select name, amount
insert into BulkOrdersStream;

Release note

Add Siddhi error handling (Stream/Store/Wait) support for Kafka Sink

Documentation

General Siddhi doc for error handling is applicable for the Kafka Sink as well.

Automation tests

• Unit tests : Add Kafka Sink Error Stream Test case #145

Security checks

• Followed secure coding standards in http://wso2.com/technical-reports/wso2-secure-engineering-guidelines? yes
• Ran FindSecurityBugs plugin and verified report? yes
• Confirmed that this PR doesn't commit any keys, passwords, tokens, usernames, or other secrets? yes

Related PRs

Below PR needs to be merged for this feature to work properly - siddhi-io/siddhi#1755

Migrations (if applicable)

None.

Test environment

SI 4.0.0
kafka_2.11-0.10.2.1
java version "1.8.0_301"
Java(TM) SE Runtime Environment (build 1.8.0_301-b09)
Java HotSpot(TM) 64-Bit Server VM (build 25.301-b09, mixed mode)