diff --git a/packages/google-cloud-bigquery-datatransfer/.coveragerc b/packages/google-cloud-bigquery-datatransfer/.coveragerc
index 38dd96f7988d..63daceafe909 100644
--- a/packages/google-cloud-bigquery-datatransfer/.coveragerc
+++ b/packages/google-cloud-bigquery-datatransfer/.coveragerc
@@ -2,7 +2,6 @@
 branch = True
 
 [report]
-fail_under = 100
 show_missing = True
 omit =
     google/cloud/bigquery_datatransfer/__init__.py
diff --git a/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/async_client.py b/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/async_client.py
index d0454623b0de..7f502338d0c2 100644
--- a/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/async_client.py
+++ b/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/async_client.py
@@ -18,6 +18,7 @@
 import re
 from typing import Dict, Sequence, Tuple, Type, Union
 import pkg_resources
+import warnings
 
 import google.api_core.client_options as ClientOptions  # type: ignore
 from google.api_core import exceptions as core_exceptions  # type: ignore
@@ -861,6 +862,11 @@ async def schedule_transfer_runs(
                 for a time range.
 
         """
+        warnings.warn(
+            "DataTransferServiceAsyncClient.schedule_transfer_runs is deprecated",
+            DeprecationWarning,
+        )
+
         # Create or coerce a protobuf request object.
         # Sanity check: If we got a request object, we should *not* have
         # gotten any keyword arguments that map to the request.
diff --git a/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/client.py b/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/client.py
index b0dab00fbb76..099e611b27e1 100644
--- a/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/client.py
+++ b/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/client.py
@@ -19,6 +19,7 @@
 import re
 from typing import Callable, Dict, Optional, Sequence, Tuple, Type, Union
 import pkg_resources
+import warnings
 
 from google.api_core import client_options as client_options_lib  # type: ignore
 from google.api_core import exceptions as core_exceptions  # type: ignore
@@ -1020,6 +1021,11 @@ def schedule_transfer_runs(
                 for a time range.
 
         """
+        warnings.warn(
+            "DataTransferServiceClient.schedule_transfer_runs is deprecated",
+            DeprecationWarning,
+        )
+
         # Create or coerce a protobuf request object.
         # Sanity check: If we got a request object, we should *not* have
         # gotten any keyword arguments that map to the request.
diff --git a/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/base.py b/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/base.py
index 830861c5ec83..d4a0ff105153 100644
--- a/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/base.py
+++ b/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/base.py
@@ -24,6 +24,7 @@
 from google.api_core import gapic_v1  # type: ignore
 from google.api_core import retry as retries  # type: ignore
 from google.auth import credentials as ga_credentials  # type: ignore
+from google.oauth2 import service_account  # type: ignore
 
 from google.cloud.bigquery_datatransfer_v1.types import datatransfer
 from google.cloud.bigquery_datatransfer_v1.types import transfer
@@ -47,8 +48,6 @@
     except pkg_resources.DistributionNotFound:  # pragma: NO COVER
         _GOOGLE_AUTH_VERSION = None
 
-_API_CORE_VERSION = google.api_core.__version__
-
 
 class DataTransferServiceTransport(abc.ABC):
     """Abstract transport class for DataTransferService."""
@@ -66,6 +65,7 @@ def __init__(
         scopes: Optional[Sequence[str]] = None,
         quota_project_id: Optional[str] = None,
         client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+        always_use_jwt_access: Optional[bool] = False,
         **kwargs,
     ) -> None:
         """Instantiate the transport.
@@ -89,6 +89,8 @@ def __init__(
                 API requests. If ``None``, then default info will be used.
                 Generally, you only need to set this if you're developing
                 your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
         """
         # Save the hostname. Default to port 443 (HTTPS) if none is specified.
         if ":" not in host:
@@ -98,7 +100,7 @@ def __init__(
         scopes_kwargs = self._get_scopes_kwargs(self._host, scopes)
 
         # Save the scopes.
-        self._scopes = scopes or self.AUTH_SCOPES
+        self._scopes = scopes
 
         # If no credentials are provided, then determine the appropriate
         # defaults.
@@ -117,13 +119,20 @@ def __init__(
                 **scopes_kwargs, quota_project_id=quota_project_id
             )
 
+        # If the credentials is service account credentials, then always try to use self signed JWT.
+        if (
+            always_use_jwt_access
+            and isinstance(credentials, service_account.Credentials)
+            and hasattr(service_account.Credentials, "with_always_use_jwt_access")
+        ):
+            credentials = credentials.with_always_use_jwt_access(True)
+
         # Save the credentials.
         self._credentials = credentials
 
-    # TODO(busunkim): These two class methods are in the base transport
+    # TODO(busunkim): This method is in the base transport
     # to avoid duplicating code across the transport classes. These functions
-    # should be deleted once the minimum required versions of google-api-core
-    # and google-auth are increased.
+    # should be deleted once the minimum required versions of google-auth is increased.
 
     # TODO: Remove this function once google-auth >= 1.25.0 is required
     @classmethod
@@ -144,27 +153,6 @@ def _get_scopes_kwargs(
 
         return scopes_kwargs
 
-    # TODO: Remove this function once google-api-core >= 1.26.0 is required
-    @classmethod
-    def _get_self_signed_jwt_kwargs(
-        cls, host: str, scopes: Optional[Sequence[str]]
-    ) -> Dict[str, Union[Optional[Sequence[str]], str]]:
-        """Returns kwargs to pass to grpc_helpers.create_channel depending on the google-api-core version"""
-
-        self_signed_jwt_kwargs: Dict[str, Union[Optional[Sequence[str]], str]] = {}
-
-        if _API_CORE_VERSION and (
-            packaging.version.parse(_API_CORE_VERSION)
-            >= packaging.version.parse("1.26.0")
-        ):
-            self_signed_jwt_kwargs["default_scopes"] = cls.AUTH_SCOPES
-            self_signed_jwt_kwargs["scopes"] = scopes
-            self_signed_jwt_kwargs["default_host"] = cls.DEFAULT_HOST
-        else:
-            self_signed_jwt_kwargs["scopes"] = scopes or cls.AUTH_SCOPES
-
-        return self_signed_jwt_kwargs
-
     def _prep_wrapped_messages(self, client_info):
         # Precompute the wrapped methods.
         self._wrapped_methods = {
diff --git a/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/grpc.py b/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/grpc.py
index 9181ae482a68..6f431d2a6378 100644
--- a/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/grpc.py
+++ b/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/grpc.py
@@ -62,6 +62,7 @@ def __init__(
         client_cert_source_for_mtls: Callable[[], Tuple[bytes, bytes]] = None,
         quota_project_id: Optional[str] = None,
         client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+        always_use_jwt_access: Optional[bool] = False,
     ) -> None:
         """Instantiate the transport.
 
@@ -102,6 +103,8 @@ def __init__(
                 API requests. If ``None``, then default info will be used.
                 Generally, you only need to set this if you're developing
                 your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
 
         Raises:
           google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
@@ -154,6 +157,7 @@ def __init__(
             scopes=scopes,
             quota_project_id=quota_project_id,
             client_info=client_info,
+            always_use_jwt_access=always_use_jwt_access,
         )
 
         if not self._grpc_channel:
@@ -209,14 +213,14 @@ def create_channel(
               and ``credentials_file`` are passed.
         """
 
-        self_signed_jwt_kwargs = cls._get_self_signed_jwt_kwargs(host, scopes)
-
         return grpc_helpers.create_channel(
             host,
             credentials=credentials,
             credentials_file=credentials_file,
             quota_project_id=quota_project_id,
-            **self_signed_jwt_kwargs,
+            default_scopes=cls.AUTH_SCOPES,
+            scopes=scopes,
+            default_host=cls.DEFAULT_HOST,
             **kwargs,
         )
 
diff --git a/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/grpc_asyncio.py b/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/grpc_asyncio.py
index 2cd986db0661..a5f7d8f4e34b 100644
--- a/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/grpc_asyncio.py
+++ b/packages/google-cloud-bigquery-datatransfer/google/cloud/bigquery_datatransfer_v1/services/data_transfer_service/transports/grpc_asyncio.py
@@ -83,14 +83,14 @@ def create_channel(
             aio.Channel: A gRPC AsyncIO channel object.
         """
 
-        self_signed_jwt_kwargs = cls._get_self_signed_jwt_kwargs(host, scopes)
-
         return grpc_helpers_async.create_channel(
             host,
             credentials=credentials,
             credentials_file=credentials_file,
             quota_project_id=quota_project_id,
-            **self_signed_jwt_kwargs,
+            default_scopes=cls.AUTH_SCOPES,
+            scopes=scopes,
+            default_host=cls.DEFAULT_HOST,
             **kwargs,
         )
 
@@ -108,6 +108,7 @@ def __init__(
         client_cert_source_for_mtls: Callable[[], Tuple[bytes, bytes]] = None,
         quota_project_id=None,
         client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+        always_use_jwt_access: Optional[bool] = False,
     ) -> None:
         """Instantiate the transport.
 
@@ -149,6 +150,8 @@ def __init__(
                 API requests. If ``None``, then default info will be used.
                 Generally, you only need to set this if you're developing
                 your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
 
         Raises:
             google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport
@@ -200,6 +203,7 @@ def __init__(
             scopes=scopes,
             quota_project_id=quota_project_id,
             client_info=client_info,
+            always_use_jwt_access=always_use_jwt_access,
         )
 
         if not self._grpc_channel:
diff --git a/packages/google-cloud-bigquery-datatransfer/owlbot.py b/packages/google-cloud-bigquery-datatransfer/owlbot.py
index 2281a7a80e9a..6a5c03b16d73 100644
--- a/packages/google-cloud-bigquery-datatransfer/owlbot.py
+++ b/packages/google-cloud-bigquery-datatransfer/owlbot.py
@@ -27,17 +27,12 @@
 # ----------------------------------------------------------------------------
 for library in s.get_staging_dirs("v1"):
 
-    # Comment out broken assertion in unit test
-    # https://github.com/googleapis/gapic-generator-python/issues/897
+    # Fix incorrect DeprecationWarning
+    # Fixed in https://github.com/googleapis/gapic-generator-python/pull/943
     s.replace(
-        library / "tests/**/*.py",
-        "assert args\[0\]\.start_time == timestamp_pb2\.Timestamp\(seconds=751\)",
-        "# assert args[0].start_time == timestamp_pb2.Timestamp(seconds=751)"
-    )
-    s.replace(
-        library / "tests/**/*.py",
-        "assert args\[0\]\.end_time == timestamp_pb2\.Timestamp\(seconds=751\)",
-        "# assert args[0].end_time == timestamp_pb2.Timestamp(seconds=751)"
+        "google/**/*client.py",
+        "warnings\.DeprecationWarning",
+        "DeprecationWarning"
     )
 
     s.move(library, excludes=["*.tar.gz", "docs/index.rst", "README.rst", "setup.py"])
diff --git a/packages/google-cloud-bigquery-datatransfer/setup.py b/packages/google-cloud-bigquery-datatransfer/setup.py
index 043dd1790083..d1ac14ca9783 100644
--- a/packages/google-cloud-bigquery-datatransfer/setup.py
+++ b/packages/google-cloud-bigquery-datatransfer/setup.py
@@ -29,7 +29,7 @@
 # 'Development Status :: 5 - Production/Stable'
 release_status = "Development Status :: 5 - Production/Stable"
 dependencies = (
-    "google-api-core[grpc] >= 1.22.2, < 2.0.0dev",
+    "google-api-core[grpc] >= 1.26.0, <2.0.0dev",
     "proto-plus >= 1.15.0",
     "packaging >= 14.3",
 )
diff --git a/packages/google-cloud-bigquery-datatransfer/testing/constraints-3.6.txt b/packages/google-cloud-bigquery-datatransfer/testing/constraints-3.6.txt
index e6739c3e2098..d3b6361ca137 100644
--- a/packages/google-cloud-bigquery-datatransfer/testing/constraints-3.6.txt
+++ b/packages/google-cloud-bigquery-datatransfer/testing/constraints-3.6.txt
@@ -5,7 +5,7 @@
 #
 # e.g., if setup.py has "foo >= 1.14.0, < 2.0.0dev",
 # Then this file should have foo==1.14.0
-google-api-core==1.22.2
+google-api-core==1.26.0
 proto-plus==1.15.0
 libcst==0.2.5
 packaging==14.3
diff --git a/packages/google-cloud-bigquery-datatransfer/tests/unit/gapic/bigquery_datatransfer_v1/test_data_transfer_service.py b/packages/google-cloud-bigquery-datatransfer/tests/unit/gapic/bigquery_datatransfer_v1/test_data_transfer_service.py
index cf70ccc92776..be3b2c25315e 100644
--- a/packages/google-cloud-bigquery-datatransfer/tests/unit/gapic/bigquery_datatransfer_v1/test_data_transfer_service.py
+++ b/packages/google-cloud-bigquery-datatransfer/tests/unit/gapic/bigquery_datatransfer_v1/test_data_transfer_service.py
@@ -41,9 +41,6 @@
 from google.cloud.bigquery_datatransfer_v1.services.data_transfer_service import (
     transports,
 )
-from google.cloud.bigquery_datatransfer_v1.services.data_transfer_service.transports.base import (
-    _API_CORE_VERSION,
-)
 from google.cloud.bigquery_datatransfer_v1.services.data_transfer_service.transports.base import (
     _GOOGLE_AUTH_VERSION,
 )
@@ -58,8 +55,9 @@
 import google.auth
 
 
-# TODO(busunkim): Once google-api-core >= 1.26.0 is required:
-# - Delete all the api-core and auth "less than" test cases
+# TODO(busunkim): Once google-auth >= 1.25.0 is required transitively
+# through google-api-core:
+# - Delete the auth "less than" test cases
 # - Delete these pytest markers (Make the "greater than or equal to" tests the default).
 requires_google_auth_lt_1_25_0 = pytest.mark.skipif(
     packaging.version.parse(_GOOGLE_AUTH_VERSION) >= packaging.version.parse("1.25.0"),
@@ -70,16 +68,6 @@
     reason="This test requires google-auth >= 1.25.0",
 )
 
-requires_api_core_lt_1_26_0 = pytest.mark.skipif(
-    packaging.version.parse(_API_CORE_VERSION) >= packaging.version.parse("1.26.0"),
-    reason="This test requires google-api-core < 1.26.0",
-)
-
-requires_api_core_gte_1_26_0 = pytest.mark.skipif(
-    packaging.version.parse(_API_CORE_VERSION) < packaging.version.parse("1.26.0"),
-    reason="This test requires google-api-core >= 1.26.0",
-)
-
 
 def client_cert_source_callback():
     return b"cert bytes", b"key bytes"
@@ -143,6 +131,36 @@ def test_data_transfer_service_client_from_service_account_info(client_class):
         assert client.transport._host == "bigquerydatatransfer.googleapis.com:443"
 
 
+@pytest.mark.parametrize(
+    "client_class", [DataTransferServiceClient, DataTransferServiceAsyncClient,]
+)
+def test_data_transfer_service_client_service_account_always_use_jwt(client_class):
+    with mock.patch.object(
+        service_account.Credentials, "with_always_use_jwt_access", create=True
+    ) as use_jwt:
+        creds = service_account.Credentials(None, None, None)
+        client = client_class(credentials=creds)
+        use_jwt.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "transport_class,transport_name",
+    [
+        (transports.DataTransferServiceGrpcTransport, "grpc"),
+        (transports.DataTransferServiceGrpcAsyncIOTransport, "grpc_asyncio"),
+    ],
+)
+def test_data_transfer_service_client_service_account_always_use_jwt_true(
+    transport_class, transport_name
+):
+    with mock.patch.object(
+        service_account.Credentials, "with_always_use_jwt_access", create=True
+    ) as use_jwt:
+        creds = service_account.Credentials(None, None, None)
+        transport = transport_class(credentials=creds, always_use_jwt_access=True)
+        use_jwt.assert_called_once_with(True)
+
+
 @pytest.mark.parametrize(
     "client_class", [DataTransferServiceClient, DataTransferServiceAsyncClient,]
 )
@@ -2776,8 +2794,12 @@ def test_schedule_transfer_runs_flattened():
         assert len(call.mock_calls) == 1
         _, args, _ = call.mock_calls[0]
         assert args[0].parent == "parent_value"
-        # assert args[0].start_time == timestamp_pb2.Timestamp(seconds=751)
-        # assert args[0].end_time == timestamp_pb2.Timestamp(seconds=751)
+        assert TimestampRule().to_proto(args[0].start_time) == timestamp_pb2.Timestamp(
+            seconds=751
+        )
+        assert TimestampRule().to_proto(args[0].end_time) == timestamp_pb2.Timestamp(
+            seconds=751
+        )
 
 
 def test_schedule_transfer_runs_flattened_error():
@@ -2825,8 +2847,12 @@ async def test_schedule_transfer_runs_flattened_async():
         assert len(call.mock_calls)
         _, args, _ = call.mock_calls[0]
         assert args[0].parent == "parent_value"
-        # assert args[0].start_time == timestamp_pb2.Timestamp(seconds=751)
-        # assert args[0].end_time == timestamp_pb2.Timestamp(seconds=751)
+        assert TimestampRule().to_proto(args[0].start_time) == timestamp_pb2.Timestamp(
+            seconds=751
+        )
+        assert TimestampRule().to_proto(args[0].end_time) == timestamp_pb2.Timestamp(
+            seconds=751
+        )
 
 
 @pytest.mark.asyncio
@@ -4684,7 +4710,6 @@ def test_data_transfer_service_transport_auth_adc_old_google_auth(transport_clas
         (transports.DataTransferServiceGrpcAsyncIOTransport, grpc_helpers_async),
     ],
 )
-@requires_api_core_gte_1_26_0
 def test_data_transfer_service_transport_create_channel(transport_class, grpc_helpers):
     # If credentials and host are not provided, the transport class should use
     # ADC credentials.
@@ -4713,79 +4738,6 @@ def test_data_transfer_service_transport_create_channel(transport_class, grpc_he
         )
 
 
-@pytest.mark.parametrize(
-    "transport_class,grpc_helpers",
-    [
-        (transports.DataTransferServiceGrpcTransport, grpc_helpers),
-        (transports.DataTransferServiceGrpcAsyncIOTransport, grpc_helpers_async),
-    ],
-)
-@requires_api_core_lt_1_26_0
-def test_data_transfer_service_transport_create_channel_old_api_core(
-    transport_class, grpc_helpers
-):
-    # If credentials and host are not provided, the transport class should use
-    # ADC credentials.
-    with mock.patch.object(
-        google.auth, "default", autospec=True
-    ) as adc, mock.patch.object(
-        grpc_helpers, "create_channel", autospec=True
-    ) as create_channel:
-        creds = ga_credentials.AnonymousCredentials()
-        adc.return_value = (creds, None)
-        transport_class(quota_project_id="octopus")
-
-        create_channel.assert_called_with(
-            "bigquerydatatransfer.googleapis.com:443",
-            credentials=creds,
-            credentials_file=None,
-            quota_project_id="octopus",
-            scopes=("https://www.googleapis.com/auth/cloud-platform",),
-            ssl_credentials=None,
-            options=[
-                ("grpc.max_send_message_length", -1),
-                ("grpc.max_receive_message_length", -1),
-            ],
-        )
-
-
-@pytest.mark.parametrize(
-    "transport_class,grpc_helpers",
-    [
-        (transports.DataTransferServiceGrpcTransport, grpc_helpers),
-        (transports.DataTransferServiceGrpcAsyncIOTransport, grpc_helpers_async),
-    ],
-)
-@requires_api_core_lt_1_26_0
-def test_data_transfer_service_transport_create_channel_user_scopes(
-    transport_class, grpc_helpers
-):
-    # If credentials and host are not provided, the transport class should use
-    # ADC credentials.
-    with mock.patch.object(
-        google.auth, "default", autospec=True
-    ) as adc, mock.patch.object(
-        grpc_helpers, "create_channel", autospec=True
-    ) as create_channel:
-        creds = ga_credentials.AnonymousCredentials()
-        adc.return_value = (creds, None)
-
-        transport_class(quota_project_id="octopus", scopes=["1", "2"])
-
-        create_channel.assert_called_with(
-            "bigquerydatatransfer.googleapis.com:443",
-            credentials=creds,
-            credentials_file=None,
-            quota_project_id="octopus",
-            scopes=["1", "2"],
-            ssl_credentials=None,
-            options=[
-                ("grpc.max_send_message_length", -1),
-                ("grpc.max_receive_message_length", -1),
-            ],
-        )
-
-
 @pytest.mark.parametrize(
     "transport_class",
     [
@@ -4810,7 +4762,7 @@ def test_data_transfer_service_grpc_transport_client_cert_source_for_mtls(
             "squid.clam.whelk:443",
             credentials=cred,
             credentials_file=None,
-            scopes=("https://www.googleapis.com/auth/cloud-platform",),
+            scopes=None,
             ssl_credentials=mock_ssl_channel_creds,
             quota_project_id=None,
             options=[
@@ -4919,7 +4871,7 @@ def test_data_transfer_service_transport_channel_mtls_with_client_cert_source(
                 "mtls.squid.clam.whelk:443",
                 credentials=cred,
                 credentials_file=None,
-                scopes=("https://www.googleapis.com/auth/cloud-platform",),
+                scopes=None,
                 ssl_credentials=mock_ssl_cred,
                 quota_project_id=None,
                 options=[
@@ -4966,7 +4918,7 @@ def test_data_transfer_service_transport_channel_mtls_with_adc(transport_class):
                 "mtls.squid.clam.whelk:443",
                 credentials=mock_cred,
                 credentials_file=None,
-                scopes=("https://www.googleapis.com/auth/cloud-platform",),
+                scopes=None,
                 ssl_credentials=mock_ssl_cred,
                 quota_project_id=None,
                 options=[