### 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)

self.pubsub_hook.delete_subscription(project_id=TEST_PROJECT, subscription=TEST_SUBSCRIPTION)
        delete_method = mock_service.delete_subscription
        delete_method.assert_called_once_with(
            request=dict(subscription=EXPANDED_SUBSCRIPTION), retry=DEFAULT, timeout=None, metadata=()
        )
    @mock.patch(PUBSUB_STRING.format("PubSubHook.subscriber_client"))
    def test_delete_nonexisting_subscription_failifnotexists(self, mock_service):
        mock_service.delete_subscription.side_effect = NotFound(
            f"Subscription does not exists: {EXPANDED_SUBSCRIPTION}"
        )
        with pytest.raises(PubSubException) as ctx:
            self.pubsub_hook.delete_subscription(
                project_id=TEST_PROJECT, subscription=TEST_SUBSCRIPTION, fail_if_not_exists=True
            )
        assert str(ctx.value) == f"Subscription does not exist: {EXPANDED_SUBSCRIPTION}"
    @mock.patch(PUBSUB_STRING.format("PubSubHook.subscriber_client"))
    def test_delete_subscription_api_call_error(self, mock_service):
        mock_service.delete_subscription.side_effect = GoogleAPICallError(
            f"Error deleting subscription {EXPANDED_SUBSCRIPTION}"
        )
        with pytest.raises(PubSubException):
            self.pubsub_hook.delete_subscription(
                project_id=TEST_PROJECT, subscription=TEST_SUBSCRIPTION, fail_if_not_exists=True
            )
    @mock.patch(PUBSUB_STRING.format("PubSubHook.subscriber_client"))
    @mock.patch(PUBSUB_STRING.format("uuid4"), new_callable=mock.Mock(return_value=lambda: TEST_UUID))
    def test_create_subscription_without_subscription_name(self, mock_uuid, mock_service):