fix(bigtable): Added rst_stream exception handling for ReadRows.

[daniel-sanche]

Migrating over @gkevinzheng PR from bigtable monorepo googleapis/python-bigtable#1298

Original description:

According to go/rst_stream, INTERNAL errors with error messages related to an rst_stream error should be interpreted as UNAVAILABLE errors instead of internal errors. This PR creates a custom retry predicate to allow retrying of INTERNAL errors with rst_stream specific error messages if the ServiceUnavailable exception is allowed to be retried.

Note to reviewers: This PR has already been reviewed and merged to a staging branch, with the intention of doing a single merge to main. We are now planning to slowly rollout these changes back to the main branch. Minimal re-review should be necessary

[gemini-code-assist]

Code Review

This pull request introduces a custom retry predicate, _rst_stream_aware_predicate, for ReadRows operations in both the asynchronous and synchronous Bigtable clients. This predicate treats specific InternalServerError messages (such as "rst_stream") as ServiceUnavailable errors to allow them to be retried. The review feedback points out a potential AttributeError in the predicate's implementation if e.message is None, and suggests adding a type check to ensure e.message is a string before calling .lower().

[gemini-code-assist]

If e.message is None (which can happen if an InternalServerError is instantiated without a message or with None), calling e.message.lower() will raise an AttributeError. To ensure robust defensive programming and prevent potential crashes in the retry loop, we should verify that e.message is a string before performing string operations on it.

Suggested change

	def rst_check(e):
	return (
	core_exceptions.ServiceUnavailable in exception_types
	and isinstance(e, core_exceptions.InternalServerError)
	and any(m in e.message.lower() for m in _RETRYABLE_INTERNAL_ERROR_MESSAGES)
	)
	def rst_check(e):
	return (
	core_exceptions.ServiceUnavailable in exception_types
	and isinstance(e, core_exceptions.InternalServerError)
	and isinstance(e.message, str)
	and any(m in e.message.lower() for m in _RETRYABLE_INTERNAL_ERROR_MESSAGES)
	)