### Description
Re: Google Cloud Provider -- PubSub

The PubSub [Topics Create API](https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/create#request-body) supports the SchemaSettings and messageRetentionDuration parameters. These should also be supported via the [PubSubCreateTopicOperator](https://github.com/apache/airflow/blob/9207e7d5e5c183d2e63c3030216b14709257668e/airflow/providers/google/cloud/operators/pubsub.py#L50). Thanks!

### Use case/motivation

_No response_

### Related issues

_No response_

### Are you willing to submit a PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

PubSubHook._validate_messages(messages)
    @pytest.mark.parametrize(
        "messages, error_message",
        [
            ([("wrong type",)], "Wrong message type. Must be a dictionary."),
            ([{"wrong_key": b"test"}], "Wrong message. Dictionary must contain 'data' or 'attributes'."),
            ([{"data": "wrong string"}], "Wrong message. 'data' must be send as a bytestring"),
            ([{"data": None}], "Wrong message. 'data' must be send as a bytestring"),
            (
                [{"attributes": None}],
                "Wrong message. If 'data' is not provided 'attributes' must be a non empty dictionary.",
            ),
            (
                [{"attributes": "wrong string"}],
                "Wrong message. If 'data' is not provided 'attributes' must be a non empty dictionary.",
            ),
        ],
    )
    def test_messages_validation_negative(self, messages, error_message):
        with pytest.raises(PubSubException) as ctx:
            PubSubHook._validate_messages(messages)
        assert str(ctx.value) == error_message
class TestPubSubAsyncHook: