The part of pushing records to elastic search for DSS and Collection indexes is done through Kafka-connect. The commands to create connector for indexing are as follows:-
DSS Indexing (to index
dss-collection_v2)
curl -X POST http://kafka-connect.kafka-cluster:8083/connectors/ -H 'Content-Type: application/json' -H 'Cookie: SESSIONID=f1349448-761e-4ebc-a8bb-f6799e756185' -H 'Postman-Token: adabf0e8-0599-4ac9-a591-920586ff4d50' -H 'cache-control: no-cache' -d '{"name": "cms-case-es-sink9132","config": {"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector","connection.url": "http://elasticsearch-data-v1.es-cluster:9200","type.name": "general","topics": "egov-dss-ingest-enriched","key.ignore": "false","schema.ignore": true,"value.converter.schemas.enable": false,"key.converter": "org.apache.kafka.connect.storage.StringConverter","value.converter": "org.apache.kafka.connect.json.JsonConverter","transforms": "TopicNameRouter","transforms.TopicNameRouter.type": "org.apache.kafka.connect.transforms.RegexRouter","transforms.TopicNameRouter.regex": ".*","transforms.TopicNameRouter.replacement": "dss-collection_v2","batch.size": 10,"max.buffered.records": 500,"flush.timeout.ms": 600000,"retry.backoff.ms": 5000,"read.timout.ms": 10000,"linger.ms": 100,"max.in.flight.requests": 2,"errors.log.enable": true,"errors.deadletterqueue.topic.name": "dss-collection_v2-es-failed","tasks.max": 1}}'
2. Collection Indexing (to index paymentsindex-v1):-
curl -X POST \
http://kafka-connect.kafka-cluster:8083/connectors/ \
H 'Content-Type: application/json' \
H 'Cookie: SESSIONID=f1349448-761e-4ebc-a8bb-f6799e756185' \
H 'Postman-Token: adabf0e8-0599-4ac9-a591-920586ff4d50' \
H 'cache-control: no-cache' \
d '{
"name": "cms-case-es-sink9121",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"connection.url": "http://elasticsearch-data-v1.es-cluster:9200",
"type.name": "payments",
"topics": "paymentsindex-v1-enriched",
"key.ignore": "false",
"schema.ignore": true,
"value.converter.schemas.enable": false,
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"transforms": "TopicNameRouter",
"transforms.TopicNameRouter.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.TopicNameRouter.regex": ".*",
"transforms.TopicNameRouter.replacement": "paymentsindex-v1",
"batch.size": 10,
"max.buffered.records": 500,
"flush.timeout.ms": 600000,
"retry.backoff.ms": 5000,
"read.timout.ms": 10000,
"linger.ms": 100,
"max.in.flight.requests": 2,
"errors.log.enable": true,
"errors.deadletterqueue.topic.name": "paymentsindex-v1-es-failed",
"tasks.max": 1
}
}'
3. Grievance Indexing (to index pgr-services):-
curl -X POST \
http://kafka-connect.kafka-cluster:8083/connectors/ \
H 'Content-Type: application/json' \
H 'Cookie: SESSIONID=f1349448-761e-4ebc-a8bb-f6799e756185' \
H 'Postman-Token: adabf0e8-0599-4ac9-a591-920586ff4d50' \
H 'cache-control: no-cache' \
d '{
"name": "cms-case-es-sink9121",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"connection.url": "http://elasticsearch-data-v1.es-cluster:9200",
"type.name": "general",
"topics": "pgr-services",
"key.ignore": "false",
"schema.ignore": true,
"value.converter.schemas.enable": false,
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"transforms": "TopicNameRouter",
"transforms.TopicNameRouter.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.TopicNameRouter.regex": ".*",
"transforms.TopicNameRouter.replacement": "pgr-services",
"batch.size": 10,
"max.buffered.records": 500,
"flush.timeout.ms": 600000,
"retry.backoff.ms": 5000,
"read.timout.ms": 10000,
"linger.ms": 100,
"max.in.flight.requests": 2,
"errors.log.enable": true,
"errors.deadletterqueue.topic.name": "pgr-services-es-failed",
"tasks.max": 1
}
}'