Bug in AWS Neptune-to-OpenSearch replication CloudFormation (missing import es_helper in replication_handler.py) — eu-west-1

Question

I am encountering a bug in the AWS provided Neptune-to-OpenSearch replication. Specifically in the NeptuneStreamPollerLambda. A reference is used to `es_helper`, while it isn't imported in the script. I have tracked the issue down to the replication_helper.py in the neptune-streams-layer.zip in this bucket: s3://aws-neptune-customer-samples-us-east-1/neptune-stream/lambda/python39/release_2023_09_04/

On line 328 there is a reference to `es_helper.get_url_components(` while `es_helper` is not imported.

The complete stacktrace of the error is:

```
[ERROR] NameError: name 'es_helper' is not defined
Traceback (most recent call last):
  File "/opt/python/lambda_function.py", line 183, in lambda_handler
    default_handler_replication(lease, wait_time, execution_end_time)
  File "/opt/python/lambda_function.py", line 131, in default_handler_replication
    raise e
  File "/opt/python/lambda_function.py", line 113, in default_handler_replication
    if not stream_records_processor.write_with_metrics(query_queue, lease, lease_manager, metrics_publisher_client):
  File "/opt/python/stream_records_processor.py", line 162, in write_with_metrics
    result = self.write(query_queue)
  File "/opt/python/stream_records_processor.py", line 144, in write
    return stream_records_handler.write_records(query_queue.get())
  File "/opt/python/replication_handler.py", line 301, in write_records
    raise e # Raise exception to avoid updating checkpoint in case of write failure
  File "/opt/python/replication_handler.py", line 298, in write_records
    future.result() # Need to call result() so that exception can be raised
  File "/var/lang/lib/python3.9/concurrent/futures/_base.py", line 439, in result
    return self.__get_result()
  File "/var/lang/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
  File "/var/lang/lib/python3.9/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/opt/python/replication_handler.py", line 328, in __run_query__
    ES_ENDPOINT = es_helper.get_url_components(config_provider.get_handler_additional_param('ElasticSearchEndpoint'))
```
 
Is it possible for AWS to fix this issue on a short notice?
What are workarounds for me on the short term?

Dennis · Accepted Answer

It appears that the original exception is thrown because of an issue in the error handling code. This hid the root cause of the problem, being emoji / utf-16 constructs in the source data. I could only find this out after commenting out the erroneous code in the exception handler (which was hidden in a lambda layer, so pretty difficult to find and change).

I changed the following code in the NeptuneStreamPollerLambda to resolve this particular error.

```
# opensearchpy/client/actions.py line 123
cur_size += len(data.encode("utf-8")) + 1
# change to
cur_size += len(data.encode("utf-8", errors="surrogatepass")) + 1
```

I believe the string causing this error was "Kan Boer Ayoub tractor 🚜 pulling winnen? | Motor Maniacs", though I suspect any string with an emoji in it could cause this issue.

re:Post Agent · Answer

This is indeed a bug in the Neptune-to-OpenSearch replication Lambda function where there's a reference to `es_helper` that isn't imported in the replication_handler.py file.

For a short-term workaround, you have a few options:

1. Create a custom Lambda layer with the fixed code:
- Download the neptune-streams-layer.zip from the S3 bucket
- Extract it and add the missing import statement for es_helper in replication_handler.py
- Repackage the layer and upload it to your own S3 bucket
- Modify the CloudFormation template to use your custom layer instead

2. Modify the Lambda function directly:
- After deployment, go to the Lambda console
- Edit the function code to add the missing import statement
- Save the changes

3. Use a custom Lambda function implementation:
- When creating the CloudFormation stack, specify your own S3 bucket and key for the Lambda code artifacts
- Ensure your implementation includes the proper imports

When implementing the fix, you'll need to add an import statement for es_helper at the top of the replication_handler.py file.

For the Lambda runtime, you can choose either Python 3.9 or Java 8 when setting up the CloudFormation stack, so you might consider using Java 8 as an alternative if the Python implementation continues to cause issues.

Note that when using the Neptune-to-OpenSearch replication, there are some limitations to be aware of, such as blank nodes not being replicated, and if you have fine-grained access control on your OpenSearch cluster, you'll need to enable IAM authentication in your Neptune database as well.  
**Sources**  
[Neptune exporter tool to S3 running on Windows EC2 | AWS re:Post](https://repost.aws/questions/QUvuCvA3hUQGGifKybR9Mf9g/neptune-exporter-tool-to-s3-running-on-windows-ec2)  
[Lambda layer: "Unable to import module: No module named 'opensearchpy'" | AWS re:Post](https://repost.aws/questions/QUPHB8rlEMSHaV0jixhlNwNw/lambda-layer-unable-to-import-module-no-module-named-opensearchpy)  
[Full text search in Amazon Neptune using Amazon OpenSearch Service - Amazon Neptune](https://docs.aws.amazon.com/neptune/latest/userguide/full-text-search.html)  
[Using an AWS CloudFormation template to start Neptune-to-OpenSearch replication - Amazon Neptune](https://docs.aws.amazon.com/neptune/latest/userguide/full-text-search-cfn-create.html)

Bug in AWS Neptune-to-OpenSearch replication CloudFormation (missing import es_helper in replication_handler.py) — eu-west-1

Add your answer

Relevant content