11.master_registration_center.api

# these following will be taken from env
#DB_USER=
#DB_PORT=
#DB_HOSTNAME=
#DB_PASS=
#DB_PREFIX_INDEX=
#ES_URL=

CONN_NAME="master_registration_center_$DB_PREFIX_INDEX"; # change this.. give unique name for each db/table

ES_CONN_URL='es-connect:8083'; # needn't change .. this is the ES-connector service name

DB_NAME='mosip_master'; # change this
DB_TABLES='master.registration_center'
ES_INDICES=$(echo $DB_TABLES | sed -E "s/([^,]+)/$DB_PREFIX_INDEX.\1/g")

curl \
  -X POST \
  http://$ES_CONN_URL/connectors \
  -H 'Content-Type: application/json' \
  -d \
  '{
      "name": '\"$CONN_NAME\"',
      "config": {
          "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
          "name": '\"$CONN_NAME\"',
          "connection.url": '\"$ES_URL\"',
          "tasks.max": "1",
          "topics": '\"$ES_INDICES\"',
          "key.ignore": "false",
          "schema.ignore": "true",
          "key.converter": "org.apache.kafka.connect.storage.StringConverter",
          "value.converter": "org.apache.kafka.connect.json.JsonConverter",
          "key.converter.schemas.enable": "false",
          "value.converter.schemas.enable": "true",

          "behavior.on.null.values": "DELETE",

          "transforms": "debezExtract,tsconvert01,tsconvert02,tsconvert03,tsconvert04,tsconvert05,tsconvert06,tsconvert07,tsconvert08,tsconvert09,tsSelect",

          "transforms.debezExtract.type": "io.debezium.transforms.ExtractNewRecordState",
          "transforms.debezExtract.delete.handling.mode": "drop",
          "transforms.debezExtract.drop.tombstones": "false",
          "transforms.debezExtract.add.fields": "source.ts_ms:ts_ms,table,lsn",
          "transforms.debezExtract.add.fields.prefix": "source_",

          "transforms.tsconvert01.type": "io.mosip.kafka.connect.transforms.TimestampConverterAdv$Value",
          "transforms.tsconvert01.field": "source_ts_ms",

          "transforms.tsconvert02.type": "io.mosip.kafka.connect.transforms.TimestampConverterAdv$Value",
          "transforms.tsconvert02.field": "cr_dtimes",
          "transforms.tsconvert02.input.type": "micro_sec",

          "transforms.tsconvert03.type": "io.mosip.kafka.connect.transforms.TimestampConverterAdv$Value",
          "transforms.tsconvert03.field": "upd_dtimes",
          "transforms.tsconvert03.input.type": "micro_sec",

          "transforms.tsconvert04.type": "io.mosip.kafka.connect.transforms.TimestampConverterAdv$Value",
          "transforms.tsconvert04.field": "del_dtimes",
          "transforms.tsconvert04.input.type": "micro_sec",

          "transforms.tsconvert05.type": "io.mosip.kafka.connect.transforms.TimestampConverterAdv$Value",
          "transforms.tsconvert05.field": "per_kiosk_process_time",
          "transforms.tsconvert05.input.type": "micro_sec",
          "transforms.tsconvert05.output.format": "HH:mm:ss",

          "transforms.tsconvert06.type": "io.mosip.kafka.connect.transforms.TimestampConverterAdv$Value",
          "transforms.tsconvert06.field": "center_start_time",
          "transforms.tsconvert06.input.type": "micro_sec",
          "transforms.tsconvert06.output.format": "HH:mm:ss",

          "transforms.tsconvert07.type": "io.mosip.kafka.connect.transforms.TimestampConverterAdv$Value",
          "transforms.tsconvert07.field": "center_end_time",
          "transforms.tsconvert07.input.type": "micro_sec",
          "transforms.tsconvert07.output.format": "HH:mm:ss",

          "transforms.tsconvert08.type": "io.mosip.kafka.connect.transforms.TimestampConverterAdv$Value",
          "transforms.tsconvert08.field": "lunch_start_time",
          "transforms.tsconvert08.input.type": "micro_sec",
          "transforms.tsconvert08.output.format": "HH:mm:ss",

          "transforms.tsconvert09.type": "io.mosip.kafka.connect.transforms.TimestampConverterAdv$Value",
          "transforms.tsconvert09.field": "lunch_end_time",
          "transforms.tsconvert09.input.type": "micro_sec",
          "transforms.tsconvert09.output.format": "HH:mm:ss",

          "transforms.tsSelect.type": "io.mosip.kafka.connect.transforms.TimestampSelector$Value",
          "transforms.tsSelect.ts.order": "upd_dtimes,cr_dtimes",
          "transforms.tsSelect.output.field": "@timestamp_gen"
      }
}';
sleep 30s;