dead letter queue for sink connector

[zzzz465]

Hi, I've found DLQ for source/sink connector is implemented in this issue #7032.
but I cannot make elasticsearch sink connector to send failed message to DLQ (malformed document).

I'm using apache pulsar v3.0.0, with es sink connector v3.0.0.nar. function was deployed with this command

bin/pulsar-admin sinks update \
  -a "https://www.apache.org/dyn/mirrors/mirrors.cgi?action=download&filename=pulsar/pulsar-3.0.0/connectors/pulsar-io-elastic-search-3.0.0.nar" \
  --name <name> \
  --tenant <tenant> \
  --namespace <namespace> \
  --topics-pattern "persistent://<topic-pattern>" \
  --max-redeliver-count 1 \
  --dead-letter-topic <dlq-topic> \
  --parallelism 1 \
  --classname org.apache.pulsar.io.elasticsearch.ElasticSearchSink \
  --custom-runtime-options '{
    "jobName": "<job-name-to-avoid-pod-name-include-dot>"
  }' \
  --sink-config '{
    "elasticSearchUrl": "<internal-es-uri>:9200",
    "indexName": "<index-name>",
    "bulkEnabled": false,
    "malformedDocAction": "WARN",
    "compatibilityMode": "ELASTICSEARCH_7",
    "compressionEnabled": false
  }'

as you see, I've enabled malformedDocAction to WARN, and pass dead-letter-topic` to use DLQ.

[tisonkun]

cc @zzzming @nicoloboschi do you have some insights here?