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

with pytest.raises(RuntimeError):
            PubSubHook(gcp_conn_id="GCP_CONN_ID", delegate_to="delegate_to")
    def setup_method(self):
        with mock.patch(BASE_STRING.format("GoogleBaseHook.__init__"), new=mock_init):
            self.pubsub_hook = PubSubHook(gcp_conn_id="test")
    @mock.patch("airflow.providers.google.cloud.hooks.pubsub.PubSubHook.get_credentials")
    @mock.patch("airflow.providers.google.cloud.hooks.pubsub.PublisherClient")
    def test_publisher_client_creation(self, mock_client, mock_get_creds):
        assert self.pubsub_hook._client is None
        result = self.pubsub_hook.get_conn()
        mock_client.assert_called_once_with(credentials=mock_get_creds.return_value, client_info=CLIENT_INFO)
        assert mock_client.return_value == result
        assert self.pubsub_hook._client == result
    @mock.patch("airflow.providers.google.cloud.hooks.pubsub.PubSubHook.get_credentials")
    @mock.patch("airflow.providers.google.cloud.hooks.pubsub.SubscriberClient")
    def test_subscriber_client_creation(self, mock_client, mock_get_creds):
        assert self.pubsub_hook._client is None
        result = self.pubsub_hook.subscriber_client
        mock_client.assert_called_once_with(credentials=mock_get_creds.return_value, client_info=CLIENT_INFO)
        assert mock_client.return_value == result
    @mock.patch(PUBSUB_STRING.format("PubSubHook.get_conn"))
    def test_create_nonexistent_topic(self, mock_service):