MySQL Connector to Elasticsearch Cloud Failing with ElasticsearchOverloadedError and Bulk Request Issues

Elastic Search
1

I'm new to elasticsearch

I’m encountering issues while syncing data from MySQL to Elasticsearch Cloud (free 15-day trial tier) using the MySQL connector. The connector is able to connect to MySQL and Elasticsearch, but the sync job keeps failing with the following error messages:

Error Logs from the Connector New Logs:

[FMWK][15:18:09][INFO] [Connector id: ADMp35MBB5sivtOKoyGn, index name: connector-mysql-74b1] Successfully connected to the MySQL Server.
[FMWK][15:18:53][WARNING] Client method 'bulk' retry 1: connection timeout
[FMWK][15:20:50][ERROR] [Connector id: ADMp35MBB5sivtOKoyGn, index name: connector-mysql-74b1, Sync job id: Rv9435MB6DbbSf4pe9a3] Extractor failed with an error: MemQueue has been full for 600.5129s. while timeout is 600s.
[FMWK][15:20:50][ERROR] [Connector id: ADMp35MBB5sivtOKoyGn, index name: connector-mysql-74b1, Sync job id: Rv9435MB6DbbSf4pe9a3] Sync was throttled by Elasticsearch
[FMWK][15:20:51][INFO] [Connector id: ADMp35MBB5sivtOKoyGn, index name: connector-mysql-74b1, Sync job id: Rv9435MB6DbbSf4pe9a3] Canceling the Sink task: Sink for JobType.FULL sync to connector-mysql-74b1

also I see this in the logs:

[FMWK][15:27:27][ERROR] Exception found for task Elasticsearch Sink: _bulk batch #2500
Traceback (most recent call last):
  File "/app/connectors/logger.py", line 264, in _awrapped
    return await func(*args, **kw)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/sink.py", line 191, in _batch_bulk
    res = await self.client.bulk_insert(operations, self.pipeline["name"])
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/management_client.py", line 277, in bulk_insert
    return await self._retrier.execute_with_retry(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/client.py", line 258, in execute_with_retry
    raise RetryInterruptedError(msg)
connectors.es.client.RetryInterruptedError: Retry operation was interrupted
[FMWK][15:27:27][ERROR] Exception found for task Elasticsearch Sink: _bulk batch #1500
Traceback (most recent call last):
  File "/app/connectors/logger.py", line 264, in _awrapped
    return await func(*args, **kw)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/sink.py", line 191, in _batch_bulk
    res = await self.client.bulk_insert(operations, self.pipeline["name"])
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/management_client.py", line 277, in bulk_insert
    return await self._retrier.execute_with_retry(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/client.py", line 258, in execute_with_retry
    raise RetryInterruptedError(msg)
connectors.es.client.RetryInterruptedError: Retry operation was interrupted
[FMWK][15:27:27][ERROR] Exception found for task Elasticsearch Sink: _bulk batch #2000
Traceback (most recent call last):
  File "/app/connectors/logger.py", line 264, in _awrapped
    return await func(*args, **kw)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/sink.py", line 191, in _batch_bulk
    res = await self.client.bulk_insert(operations, self.pipeline["name"])
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/management_client.py", line 277, in bulk_insert
    return await self._retrier.execute_with_retry(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/client.py", line 258, in execute_with_retry
    raise RetryInterruptedError(msg)
connectors.es.client.RetryInterruptedError: Retry operation was interrupted
[FMWK][15:27:27][ERROR] Exception found for task Elasticsearch Sink: _bulk batch #1000
Traceback (most recent call last):
  File "/app/connectors/logger.py", line 264, in _awrapped
    return await func(*args, **kw)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/sink.py", line 191, in _batch_bulk
    res = await self.client.bulk_insert(operations, self.pipeline["name"])
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/management_client.py", line 277, in bulk_insert
    return await self._retrier.execute_with_retry(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/client.py", line 258, in execute_with_retry
    raise RetryInterruptedError(msg)
connectors.es.client.RetryInterruptedError: Retry operation was interrupted
[FMWK][15:27:27][ERROR] Exception found for task Elasticsearch Sink: _bulk batch #500
Traceback (most recent call last):
  File "/app/connectors/logger.py", line 264, in _awrapped
    return await func(*args, **kw)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/sink.py", line 191, in _batch_bulk
    res = await self.client.bulk_insert(operations, self.pipeline["name"])
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/management_client.py", line 277, in bulk_insert
    return await self._retrier.execute_with_retry(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/connectors/es/client.py", line 258, in execute_with_retry
    raise RetryInterruptedError(msg)
connectors.es.client.RetryInterruptedError: Retry operation was interrupted

Elasicsearch UI I got this:

Sync failure: ElasticsearchOverloadedError: Connector was unable to ingest data into overloaded Elasticsearch. Make sure Elasticsearch instance is healthy, has enough resources and content index is healthy..