Setting AsyncPredictor.deserializer doesn't work

[athewsey]

Describe the bug

An AsyncPredictor (returned when deploying an async inference endpoint) exposes .serializer and .deserializer properties like a typical Predictor does... But behaviour of these properties is not consistent in async because of the way the class wraps around an internal .predictor.

Overriding a predictor's .serializer and .deserializer properties after creation is useful and expected functionality (IMO) because:

1. As far as I understand, the current de/serializer class architecture requires setting content_type/accept independently of individual requests. Therefore the only way to switch sent or received content type is by updating the de/serializer or re-creating the entire predictor.
2. In the past, not all predictor classes have supported serializer and deserializer constructor arguments (see feature: all predictors support serializer/deserializer overrides #1997): So some samples have expected to set these properties after constructing.

To reproduce

# Create some async predictor:
predictor = some_pytorch_model.deploy(
    async_inference_config=sagemaker.async_inference.AsyncInferenceConfig(
        output_path="s3://doc-example-bucket/folder",
        max_concurrent_invocations_per_instance=2,
    ),
)

# Override de/serializers (PyTorch defaults to Numpy):
predictor.serializer = sagemaker.serializers.JsonSerializer()
predictor.deserializer = sagemaker.deserializers.JsonDeserializer()

# Make a prediction:
resp = predictor.predict({ "hi": "there" })

Expected behavior

The endpoint receives a request with ContentType and Accept matching the configured serializers (application/json in the above example).

Actual behavior

Because AsyncPredictor uses its own serializer, the input request is as expected.

...But because it uses .predictor property's deserializer, the overrides do not affect the response: default NumpyDeserializer and application/x-npy Accept headers are still used.

Screenshots or logs

N/A

System information
A description of your system. Please provide:

• SageMaker Python SDK version: 2.86.2 (but checked problem seems to still affect master - see links)
• Framework name (eg. PyTorch) or algorithm (eg. KMeans): PyTorch (but should be general)
• Framework version: 1.10
• Python version: py3.8
• CPU or GPU: CPU
• Custom Docker image (Y/N): N

Additional context

Today we can work around this by overriding de/serializers on both the outer (async) and inner (sync) predictor objects, as follows:

async_predictor.serializer = JsonSerializer()
async_predictor.deserializer = JsonDeserializer()
async_predictor.predictor.serializer = async_predictor.serializer
async_predictor.predictor.deserializer = async_predictor.deserializer

...But I'd recommend a better solution would be to make AsyncPredictor.serializer and AsyncPredictor.deserializer into @propertys that just read from and write to the inner .predictor.(de)serializer?

[kmotohas]

I encountered the same issue too.
I think the _create_request_args function should be modified simply like this.

...
        if "EndpointName" not in args:
            args["EndpointName"] = self.endpoint_name  # self.predictor.endpoint_name originally

        if "Accept" not in args:
            args["Accept"] = ", ".join(self.deserializer.accept)  # self.predictor.accept originally
...

https://github.com/aws/sagemaker-python-sdk/blob/master/src/sagemaker/predictor_async.py#L186