Add missing fields in DynamoDBToS3Operator system test

[utkarsharma2]

There was a missing s3_key_prefix in the system test for DynamoDBToS3Operator and the export_time date format was not correct.

[uranusjr]

Is this change significant?

[utkarsharma2]

Yes, because in system tests we just want to make sure the DynamoDBToS3Operator is successfully able to call export_table_to_point_in_time with the minimum time window. So the exported data is small. But I'm open to any suggestions.

[utkarsharma2]

cc: @ferruzzi @o-nikolas @vincbeck

[o-nikolas]

As long as we don't get weird API behaviour with a relatively instantaneous export_time seems fine to me. I'm assuming you've tested this?

[utkarsharma2]

Yes, I have tested it on local and I have been developing using datetime.now() only, no issues so far.

[ferruzzi]

Why are we hard-coding the -3- part?

[utkarsharma2]

No reason was just following other examples in the file

I think it's good to have it since it isolates this test case from other test cases in the file. Otherwise, there can be some side-effects if two testcase uses the same prefix. WDYT?

[ferruzzi]

I can get behind that reasoning. I didn't see the other two in the github "changes" diff, so I didn't catch the pattern.

[o-nikolas]

I re-triggered the tests since the wait for ci images timed out.

[ferruzzi]

Third time's the charm? :P

[ferruzzi]

negative.

ERROR    airflow.executors.debug_executor.DebugExecutor:debug_executor.py:95 Failed to execute task: An error occurred (InvalidExportTimeException) when calling the ExportTableToPointInTime operation: Export Time is invalid.
--
Traceback (most recent call last):
File "/opt/airflow/airflow/executors/debug_executor.py", line 89, in _run_task
ti.run(job_id=ti.job_id, **params)
File "/opt/airflow/airflow/utils/session.py", line 76, in wrapper
return func(*args, session=session, **kwargs)
File "/opt/airflow/airflow/models/taskinstance.py", line 1798, in run
mark_success=mark_success, test_mode=test_mode, job_id=job_id, pool=pool, session=session
File "/opt/airflow/airflow/utils/session.py", line 73, in wrapper
return func(*args, **kwargs)
File "/opt/airflow/airflow/models/taskinstance.py", line 1485, in _run_raw_task
self._execute_task_with_callbacks(context, test_mode)
File "/opt/airflow/airflow/models/taskinstance.py", line 1636, in _execute_task_with_callbacks
result = self._execute_task(context, task_orig)
File "/opt/airflow/airflow/models/taskinstance.py", line 1704, in _execute_task
result = execute_callable(context=context)
File "/opt/airflow/airflow/providers/amazon/aws/transfers/dynamodb_to_s3.py", line 142, in execute
self._export_table_to_point_in_time()
File "/opt/airflow/airflow/providers/amazon/aws/transfers/dynamodb_to_s3.py", line 158, in _export_table_to_point_in_time
ExportFormat=self.export_format,
File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 530, in _api_call
return self._make_api_call(operation_name, kwargs)
File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 960, in _make_api_call
raise error_class(parsed_response, operation_name)
botocore.errorfactory.InvalidExportTimeException: An error occurred (InvalidExportTimeException) when calling the ExportTableToPointInTime operation: Export Time is invalid

[utkarsharma2]

@ferruzzi I'm not able to reproduce this on my local below code is working for me. I'm using breeze to run my test which is using docker image - ghcr.io/apache/airflow/main/ci/python3.7:latest

from datetime import datetime

from airflow import DAG
from airflow.providers.amazon.aws.transfers.dynamodb_to_s3 import DynamoDBToS3Operator

with DAG(
    dag_id='example_export_dynamodb',
    schedule_interval=None,
    start_date=datetime(2021, 1, 1),
    tags=['example'],
    catchup=False,
) as dag:
    dynamodb_to_s3_operator = DynamoDBToS3Operator(
        task_id="dynamodb_to_s3",
        dynamodb_table_name="test",
        s3_bucket_name="tmp9",
        file_size=4000,
        export_time=datetime.now(),
        s3_key_prefix="test"
    )

Can you please share the details of environment in which you are running tests? I'll try to reproduce this on my local.

[vincbeck]

Our environment is using Breeze in order to run system tests. We are building the last image with breeze ci-image build and then run the test using the command breeze testing tests <filename> --system amazon.

But given the exception I dont think it is question of breeze, the service Dynamodb is complaining that export_time is invalid. My guess is datetime.now() returns in some cases (depending on the timezone) a date in the future and in some cases a date in the present/past. I might recommend to use DateTime.UtcNow() instead with some buffer (e.g. exclude one week maybe?)

[potiuk]

Our environment is using Breeze in order to run system tests. We are building the last image with breeze ci-image build and then run the test using the command breeze testing tests <filename> --system amazon.

But given the exception I dont think it is question of breeze, the service Dynamodb is complaining that export_time is invalid. My guess is datetime.now() returns in some cases (depending on the timezone) a date in the future and in some cases a date in the present/past. I might recommend to use DateTime.UtcNow() instead with some buffer (e.g. exclude one week maybe?)

Oh yeah. I've seen it happen in the past.

[utkarsharma2]

@vincbeck I think the datetime.now() return date which is not timezone aware, so for me in India - IST this will always give time which is ahead of UTC and If the export API assumes UTC time, it should have failed for me every time(which it didn't), right? or I'm missing something?

[utkarsharma2]

@vincbeck @potiuk Added a PR - #31307 please review.