https://flink.apache.org/ logo
Join Slack
Powered by
Hi All, I am trying to sink in data to OpenSearch/...
# troubleshooting
m
Hi All, I am trying to sink in data to OpenSearch/Elasticsearch via flink connecter in upsert mode but it failing due to version conflict during the update on the document, is there any way I could tweak it to proceed with conflict.
Copy code
detect_noop[true]} of bulk request failed with status 409.
	at org.apache.flink.connector.opensearch.sink.OpensearchWriter.extractFailures(OpensearchWriter.java:346)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:398)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsNonBlocking(MailboxProcessor.java:383)
	at 
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.apache.flink.util.FlinkRuntimeException: Single action update
[OpenSearch exception [type=version_conflict_engine_exception, reason=[69404946617821]: version conflict, document already exists (current version [1])]]
t
Hi, what version are using for connector?
m
flink 1.16 , flink-opensearch-connector 2.5
t
try this.
Copy code
"org.apache.flink" % "flink-connector-elasticsearch7" % "1.16.1"
And then you can create 
IndexRequest
with create() to overwrite document.
Copy code
val rqst: IndexRequest = Requests.indexRequest
    .index(index)
    .routing(input.routing.orNull)
    .id(_id)
    .create(true)
    .source(..)
indexer.add(rqst)
m
it is failing when multiple threads are trying to update the same document
t
at the same time?
m
does multiple bulk request goes to elastic at once, if we have parallelism = 8 so will it takes 1000 * 8 documents or 1000 only.
In case of opensearch the retry mechanism seems not to be working
Copy code
org.apache.flink.util.FlinkRuntimeException: org.apache.flink.util.FlinkRuntimeException: Single action update {[v2-es-urtrf-dedup-2023.04.24][30540477598461], doc_as_upsert[false], doc[index {[null][null], source[{], scripted_upsert[false], detect_noop[true]} of bulk request failed with status 409.
opensearch fails in case of upsert
t
How do you know the bulk request is indexing the SAME document? Is the same 
_id
? In my case with above setting, It’s working fine for bulk upserting
m
for elasticsearch it’s working fine for opensearch it’s failing
I am creating a primarykey in my table sink, based on which I am upserting
Copy code
Sink DDL: (
  `id` INT NOT NULL,
  `state` STRING,
  `dob` STRING,
  `__sinkProcessingTime` STRING,
  CONSTRAINT `PK_id` PRIMARY KEY (`id`) NOT ENFORCED
)
COMMENT ''

WITH (
  'password' = '*****',
  'connector' = 'opensearch',
  'sink.bulk-flush.backoff.delay' = '18s',
  'connection.timeout' = '180s',
  'hosts' = '**',
  'sink.bulk-flush.backoff.max-retries' = '10',
  'index' = '{__sinkProcessingTime}',
  'socket.timeout' = '180s',
  'sink.bulk-flush.backoff.strategy' = 'EXPONENTIAL',
  'connection.request-timeout' = '180s',
  'username' = '**'
)
4 Views
Open in Slack
PreviousNext
Powered by