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

return [ReceivedMessage.to_dict(m) for m in self._generate_messages(count)]
    @mock.patch("airflow.providers.google.cloud.operators.pubsub.PubSubHook")
    def test_execute_no_messages(self, mock_hook):
        operator = PubSubPullOperator(
            task_id=TASK_ID,
            project_id=TEST_PROJECT,
            subscription=TEST_SUBSCRIPTION,
        )
        mock_hook.return_value.pull.return_value = []
        assert [] == operator.execute({})
    @mock.patch("airflow.providers.google.cloud.operators.pubsub.PubSubHook")
    def test_execute_with_ack_messages(self, mock_hook):
        operator = PubSubPullOperator(
            task_id=TASK_ID,
            project_id=TEST_PROJECT,
            subscription=TEST_SUBSCRIPTION,
            ack_messages=True,
        )
        generated_messages = self._generate_messages(5)
        generated_dicts = self._generate_dicts(5)
        mock_hook.return_value.pull.return_value = generated_messages
        assert generated_dicts == operator.execute({})
        mock_hook.return_value.acknowledge.assert_called_once_with(
            project_id=TEST_PROJECT,
            subscription=TEST_SUBSCRIPTION,
            messages=generated_messages,
        )
    @mock.patch("airflow.providers.google.cloud.operators.pubsub.PubSubHook")
    def test_execute_with_messages_callback(self, mock_hook):