Remove duplicate retry handler registration

The retry handlers that hook up to service-data-loaded
were added before botocore clients were added.  This results
in the client level retry handlers being a no-op if they were
already loaded (they use the same unique_id).

This means you can have subtly different behavior
depending on if you (or something else) calls
`session.get_service_data()`.

botocore/handlers.py

@@ -264,60 +264,6 @@ def _needs_s3_sse_customization(params, sse_member_prefix):
             sse_member_prefix + 'KeyMD5' not in params)
 
 
-# NOTE: Retries get registered in two separate places in the botocore
-# code: once when creating the client and once when you load the service
-# model from the session. While at first this handler seems unneeded, it
-# would be a breaking change for the AWS CLI to have it removed. This is
-# because it relies on the service model from the session to create commands
-# and this handler respects operation granular retry logic while the client
-# one does not. If this is ever to be removed the handler, the client
-# will have to respect per-operation level retry configuration.
-def register_retries_for_service(service_data, session,
-                                 service_name, **kwargs):
-    loader = session.get_component('data_loader')
-    endpoint_prefix = service_data.get('metadata', {}).get('endpointPrefix')
-    if endpoint_prefix is None:
-        logger.debug("Not registering retry handlers, could not endpoint "
-                     "prefix from model for service %s", service_name)
-        return
-    service_id = service_data.get('metadata', {}).get('serviceId')
-    if service_id is None:
-        raise MissingServiceIdError(service_name=service_name)
-    service_event_name = hyphenize_service_id(service_id)
-    config = _load_retry_config(loader, endpoint_prefix)
-    if not config:
-        return
-    logger.debug("Registering retry handlers for service: %s", service_name)
-    handler = retryhandler.create_retry_handler(
-        config, endpoint_prefix)
-    unique_id = 'retry-config-%s' % service_event_name
-    session.register('needs-retry.%s' % service_event_name,
-                     handler, unique_id=unique_id)
-    _register_for_operations(config, session, service_event_name)
-
-
-def _load_retry_config(loader, endpoint_prefix):
-    original_config = loader.load_data('_retry')
-    retry_config = translate.build_retry_config(
-        endpoint_prefix, original_config['retry'],
-        original_config.get('definitions', {}))
-    return retry_config
-
-
-def _register_for_operations(config, session, service_event_name):
-    # There's certainly a tradeoff for registering the retry config
-    # for the operations when the service is created.  In practice,
-    # there aren't a whole lot of per operation retry configs so
-    # this is ok for now.
-    for key in config:
-        if key == '__default__':
-            continue
-        handler = retryhandler.create_retry_handler(config, key)
-        unique_id = 'retry-config-%s-%s' % (service_event_name, key)
-        session.register('needs-retry.%s.%s' % (service_event_name, key),
-                         handler, unique_id=unique_id)
-
-
 def disable_signing(**kwargs):
     """
     This handler disables request signing by setting the signer
@@ -998,7 +944,6 @@ def inject_api_version_header_if_needed(model, params, **kwargs):
     ('needs-retry.s3.CopyObject', check_for_200_error, REGISTER_FIRST),
     ('needs-retry.s3.CompleteMultipartUpload', check_for_200_error,
      REGISTER_FIRST),
-    ('service-data-loaded', register_retries_for_service),
     ('choose-signer.cognito-identity.GetId', disable_signing),
     ('choose-signer.cognito-identity.GetOpenIdToken', disable_signing),
     ('choose-signer.cognito-identity.UnlinkIdentity', disable_signing),

tests/unit/test_handlers.py

@@ -552,59 +552,6 @@ def test_run_instances_userdata_blob(self):
                   'UserData': b64_user_data}
         self.assertEqual(params, result)
 
-    def test_register_retry_for_handlers_with_no_metadata(self):
-        no_endpoint_prefix = {'metadata': {}}
-        session = mock.Mock()
-        handlers.register_retries_for_service(service_data=no_endpoint_prefix,
-                                              session=mock.Mock(),
-                                              service_name='foo')
-        self.assertFalse(session.register.called)
-
-    def test_register_retry_for_handlers_with_no_service_id(self):
-        service_data = {
-            'metadata': {
-                'endpointPrefix': 'foo',
-            },
-        }
-        session = mock.Mock(spec=Session)
-        loader = mock.Mock(spec=Loader)
-        session.get_component.return_value = loader
-        service_name = 'foo'
-        with self.assertRaisesRegexp(MissingServiceIdError, service_name):
-            handlers.register_retries_for_service(
-                service_data=service_data,
-                session=session,
-                service_name=service_name,
-            )
-
-    def test_register_retry_handlers(self):
-        service_data = {
-            'metadata': {
-                'endpointPrefix': 'foo',
-                'serviceId': 'foo',
-            },
-        }
-        session = mock.Mock()
-        loader = mock.Mock()
-        session.get_component.return_value = loader
-        loader.load_data.return_value = {
-            'retry': {
-                '__default__': {
-                    'max_attempts': 10,
-                    'delay': {
-                        'type': 'exponential',
-                        'base': 2,
-                        'growth_factor': 5,
-                    },
-                },
-            },
-        }
-        handlers.register_retries_for_service(service_data=service_data,
-                                              session=session,
-                                              service_name='foo')
-        session.register.assert_called_with('needs-retry.foo', mock.ANY,
-                                            unique_id='retry-config-foo')
-
     def test_get_template_has_error_response(self):
         original = {'Error': {'Code': 'Message'}}
         handler_input = copy.deepcopy(original)
@@ -1054,10 +1001,12 @@ def get_handler_names(self, responses):
         return names
 
     def test_s3_special_case_is_before_other_retry(self):
+        client = self.session.create_client('s3')
         service_model = self.session.get_service_model('s3')
         operation = service_model.operation_model('CopyObject')
-        responses = self.session.emit(
+        responses = client.meta.events.emit(
             'needs-retry.s3.CopyObject',
+            request_dict={},
             response=(mock.Mock(), mock.Mock()), endpoint=mock.Mock(),
             operation=operation, attempts=1, caught_exception=None)
         # This is implementation specific, but we're trying to verify that