canal同步数据到ES出现丢失

[lyuxiaofei]

• I have searched the issues of this repository and believe that this is not a duplicate.
• I have checked the FAQ of this repository and believe that this is not a duplicate.

environment

• canal version
• mysql version

Issue Description

Steps to reproduce

Expected behaviour

Actual behaviour

If there is an exception, please attach the exception trace:

Just put your stack trace here!

watch org.elasticsearch.client.RestHighLevelClient bulk '{params, returnObj, throwExp}' -x 5 -n 100000

使用工具观察这个方法
method=org.elasticsearch.client.RestHighLevelClient.bulk location=AtExit
ts=2025-03-26 14:49:17.506; [cost=13.149003ms] result=@Arraylist[
@object[][
@BulkRequest[
REQUEST_OVERHEAD=@integer[50],
requests=@Arraylist[
@UpdateRequest[update {[
@IndexRequest[index {
........
indices=@hashset[
@string[transaction_detail],
],
timeout=@timevalue[
NSEC_PER_MSEC=@long[1000000],
MINUS_ONE=@timevalue[-1],
ZERO=@timevalue[0s],
C0=@long[1],
C1=@long[1000],
C2=@long[1000000],
C3=@long[1000000000],
C4=@long[60000000000],
C5=@long[3600000000000],
C6=@long[86400000000000],
duration=@long[1],
timeUnit=@[MINUTES],
],
waitForActiveShards=@ActiveShardCount[
ACTIVE_SHARD_COUNT_DEFAULT=@integer[-2],
ALL_ACTIVE_SHARDS=@integer[-1],
DEFAULT=@ActiveShardCount[DEFAULT],
ALL=@ActiveShardCount[ALL],
NONE=@ActiveShardCount[0],
ONE=@ActiveShardCount[1],
value=@integer[-2],
$assertionsDisabled=@boolean[true],
],
refreshPolicy=@RefreshPolicy[
NONE=@RefreshPolicy[NONE],
IMMEDIATE=@RefreshPolicy[IMMEDIATE],
WAIT_UNTIL=@RefreshPolicy[WAIT_UNTIL],
value=@string[false],
$VALUES=@RefreshPolicy[][isEmpty=false;size=3],
name=@string[NONE],
ordinal=@integer[0],
],
globalPipeline=null,
globalRouting=null,
globalIndex=null,
globalType=null,
sizeInBytes=@long[18550],
parentTaskId=@TaskId[
EMPTY_TASK_ID=@TaskId[unset],
nodeId=@string[],
id=@long[-1],
],
remoteAddress=null,
],
@RequestOptions[
DEFAULT=@RequestOptions[
DEFAULT=@RequestOptions[RequestOptions{}],
headers=@UnmodifiableRandomAccessList[isEmpty=true;size=0],
httpAsyncResponseConsumerFactory=@HeapBufferedResponseConsumerFactory[org.elasticsearch.client.HttpAsyncResponseConsumerFactory$HeapBufferedResponseConsumerFactory@17d49616],
warningsHandler=null,
],
headers=@UnmodifiableRandomAccessList[isEmpty=true;size=0],
httpAsyncResponseConsumerFactory=@HeapBufferedResponseConsumerFactory[
DEFAULT_BUFFER_LIMIT=@integer[104857600],
bufferLimit=@integer[104857600],
],
warningsHandler=null,
],
],
@BulkResponse[
ITEMS=@string[items],
ERRORS=@string[errors],
TOOK=@string[took],
INGEST_TOOK=@string[ingest_took],
NO_INGEST_TOOK=@long[-1],
responses=@BulkItemResponse[][
@BulkItemResponse[
发现问题：

1. requests 请求带的是全的数据，返回的BulkResponse会出现数据丢失。比如请求20个，返回10个；丢失的10个ES 也没有
   2.会出现requests10个，返回20个的情况

目前线上存在大批量数据丢失情况，帮忙解答下原因，是不是client-adapter.es7x-1.1.5-jar-with-dependencies.jar 有BUG

[lyuxiaofei]

已解决