Merge pull request #10684 from markdroth/retry

Retry support

25 files changed, 5090 insertions, 41 deletions

diff --git a/test/core/client_channel/BUILD b/test/core/client_channel/BUILD
index c4d12fe3a1..d430b722df 100644
--- a/test/core/client_channel/BUILD
+++ b/test/core/client_channel/BUILD
@@ -54,3 +54,14 @@ grpc_cc_test(
     ],
 )
 
+grpc_cc_test(
+    name = "status_util_test",
+    srcs = ["status_util_test.cc"],
+    language = "C++",
+    deps = [
+        "//:grpc",
+    ],
+    external_deps = [
+        "gtest",
+    ],
+)
diff --git a/test/core/client_channel/status_util_test.cc b/test/core/client_channel/status_util_test.cc
new file mode 100644
index 0000000000..f944990ad2
--- /dev/null
+++ b/test/core/client_channel/status_util_test.cc
@@ -0,0 +1,49 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "src/core/ext/filters/client_channel/status_util.h"
+
+#include <gtest/gtest.h>
+
+namespace grpc_core {
+namespace internal {
+namespace {
+
+TEST(StatusCodeSet, Basic) {
+  StatusCodeSet set;
+  EXPECT_TRUE(set.Empty());
+  EXPECT_FALSE(set.Contains(GRPC_STATUS_OK));
+  EXPECT_FALSE(set.Contains(GRPC_STATUS_UNAVAILABLE));
+  set.Add(GRPC_STATUS_OK);
+  EXPECT_FALSE(set.Empty());
+  EXPECT_TRUE(set.Contains(GRPC_STATUS_OK));
+  EXPECT_FALSE(set.Contains(GRPC_STATUS_UNAVAILABLE));
+  set.Add(GRPC_STATUS_UNAVAILABLE);
+  EXPECT_FALSE(set.Empty());
+  EXPECT_TRUE(set.Contains(GRPC_STATUS_OK));
+  EXPECT_TRUE(set.Contains(GRPC_STATUS_UNAVAILABLE));
+}
+
+}  // namespace
+}  // namespace internal
+}  // namespace grpc_core
+
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
diff --git a/test/core/end2end/end2end_nosec_tests.cc b/test/core/end2end/end2end_nosec_tests.cc
index 6318550ad8..78ddcdb8f6 100644
--- a/test/core/end2end/end2end_nosec_tests.cc
+++ b/test/core/end2end/end2end_nosec_tests.cc
@@ -118,6 +118,36 @@ extern void request_with_payload(grpc_end2end_test_config config);
 extern void request_with_payload_pre_init(void);
 extern void resource_quota_server(grpc_end2end_test_config config);
 extern void resource_quota_server_pre_init(void);
+extern void retry(grpc_end2end_test_config config);
+extern void retry_pre_init(void);
+extern void retry_cancellation(grpc_end2end_test_config config);
+extern void retry_cancellation_pre_init(void);
+extern void retry_disabled(grpc_end2end_test_config config);
+extern void retry_disabled_pre_init(void);
+extern void retry_exceeds_buffer_size_in_initial_batch(grpc_end2end_test_config config);
+extern void retry_exceeds_buffer_size_in_initial_batch_pre_init(void);
+extern void retry_exceeds_buffer_size_in_subsequent_batch(grpc_end2end_test_config config);
+extern void retry_exceeds_buffer_size_in_subsequent_batch_pre_init(void);
+extern void retry_non_retriable_status(grpc_end2end_test_config config);
+extern void retry_non_retriable_status_pre_init(void);
+extern void retry_recv_initial_metadata(grpc_end2end_test_config config);
+extern void retry_recv_initial_metadata_pre_init(void);
+extern void retry_recv_message(grpc_end2end_test_config config);
+extern void retry_recv_message_pre_init(void);
+extern void retry_server_pushback_delay(grpc_end2end_test_config config);
+extern void retry_server_pushback_delay_pre_init(void);
+extern void retry_server_pushback_disabled(grpc_end2end_test_config config);
+extern void retry_server_pushback_disabled_pre_init(void);
+extern void retry_streaming(grpc_end2end_test_config config);
+extern void retry_streaming_pre_init(void);
+extern void retry_streaming_after_commit(grpc_end2end_test_config config);
+extern void retry_streaming_after_commit_pre_init(void);
+extern void retry_streaming_succeeds_before_replay_finished(grpc_end2end_test_config config);
+extern void retry_streaming_succeeds_before_replay_finished_pre_init(void);
+extern void retry_throttled(grpc_end2end_test_config config);
+extern void retry_throttled_pre_init(void);
+extern void retry_too_many_attempts(grpc_end2end_test_config config);
+extern void retry_too_many_attempts_pre_init(void);
 extern void server_finishes_request(grpc_end2end_test_config config);
 extern void server_finishes_request_pre_init(void);
 extern void shutdown_finishes_calls(grpc_end2end_test_config config);
@@ -197,6 +227,21 @@ void grpc_end2end_tests_pre_init(void) {
   request_with_flags_pre_init();
   request_with_payload_pre_init();
   resource_quota_server_pre_init();
+  retry_pre_init();
+  retry_cancellation_pre_init();
+  retry_disabled_pre_init();
+  retry_exceeds_buffer_size_in_initial_batch_pre_init();
+  retry_exceeds_buffer_size_in_subsequent_batch_pre_init();
+  retry_non_retriable_status_pre_init();
+  retry_recv_initial_metadata_pre_init();
+  retry_recv_message_pre_init();
+  retry_server_pushback_delay_pre_init();
+  retry_server_pushback_disabled_pre_init();
+  retry_streaming_pre_init();
+  retry_streaming_after_commit_pre_init();
+  retry_streaming_succeeds_before_replay_finished_pre_init();
+  retry_throttled_pre_init();
+  retry_too_many_attempts_pre_init();
   server_finishes_request_pre_init();
   shutdown_finishes_calls_pre_init();
   shutdown_finishes_tags_pre_init();
@@ -265,6 +310,21 @@ void grpc_end2end_tests(int argc, char **argv,
     request_with_flags(config);
     request_with_payload(config);
     resource_quota_server(config);
+    retry(config);
+    retry_cancellation(config);
+    retry_disabled(config);
+    retry_exceeds_buffer_size_in_initial_batch(config);
+    retry_exceeds_buffer_size_in_subsequent_batch(config);
+    retry_non_retriable_status(config);
+    retry_recv_initial_metadata(config);
+    retry_recv_message(config);
+    retry_server_pushback_delay(config);
+    retry_server_pushback_disabled(config);
+    retry_streaming(config);
+    retry_streaming_after_commit(config);
+    retry_streaming_succeeds_before_replay_finished(config);
+    retry_throttled(config);
+    retry_too_many_attempts(config);
     server_finishes_request(config);
     shutdown_finishes_calls(config);
     shutdown_finishes_tags(config);
@@ -460,6 +520,66 @@ void grpc_end2end_tests(int argc, char **argv,
       resource_quota_server(config);
       continue;
     }
+    if (0 == strcmp("retry", argv[i])) {
+      retry(config);
+      continue;
+    }
+    if (0 == strcmp("retry_cancellation", argv[i])) {
+      retry_cancellation(config);
+      continue;
+    }
+    if (0 == strcmp("retry_disabled", argv[i])) {
+      retry_disabled(config);
+      continue;
+    }
+    if (0 == strcmp("retry_exceeds_buffer_size_in_initial_batch", argv[i])) {
+      retry_exceeds_buffer_size_in_initial_batch(config);
+      continue;
+    }
+    if (0 == strcmp("retry_exceeds_buffer_size_in_subsequent_batch", argv[i])) {
+      retry_exceeds_buffer_size_in_subsequent_batch(config);
+      continue;
+    }
+    if (0 == strcmp("retry_non_retriable_status", argv[i])) {
+      retry_non_retriable_status(config);
+      continue;
+    }
+    if (0 == strcmp("retry_recv_initial_metadata", argv[i])) {
+      retry_recv_initial_metadata(config);
+      continue;
+    }
+    if (0 == strcmp("retry_recv_message", argv[i])) {
+      retry_recv_message(config);
+      continue;
+    }
+    if (0 == strcmp("retry_server_pushback_delay", argv[i])) {
+      retry_server_pushback_delay(config);
+      continue;
+    }
+    if (0 == strcmp("retry_server_pushback_disabled", argv[i])) {
+      retry_server_pushback_disabled(config);
+      continue;
+    }
+    if (0 == strcmp("retry_streaming", argv[i])) {
+      retry_streaming(config);
+      continue;
+    }
+    if (0 == strcmp("retry_streaming_after_commit", argv[i])) {
+      retry_streaming_after_commit(config);
+      continue;
+    }
+    if (0 == strcmp("retry_streaming_succeeds_before_replay_finished", argv[i])) {
+      retry_streaming_succeeds_before_replay_finished(config);
+      continue;
+    }
+    if (0 == strcmp("retry_throttled", argv[i])) {
+      retry_throttled(config);
+      continue;
+    }
+    if (0 == strcmp("retry_too_many_attempts", argv[i])) {
+      retry_too_many_attempts(config);
+      continue;
+    }
     if (0 == strcmp("server_finishes_request", argv[i])) {
       server_finishes_request(config);
       continue;
diff --git a/test/core/end2end/end2end_tests.cc b/test/core/end2end/end2end_tests.cc
index 9d8dfd6723..fb1e61b39f 100644
--- a/test/core/end2end/end2end_tests.cc
+++ b/test/core/end2end/end2end_tests.cc
@@ -120,6 +120,36 @@ extern void request_with_payload(grpc_end2end_test_config config);
 extern void request_with_payload_pre_init(void);
 extern void resource_quota_server(grpc_end2end_test_config config);
 extern void resource_quota_server_pre_init(void);
+extern void retry(grpc_end2end_test_config config);
+extern void retry_pre_init(void);
+extern void retry_cancellation(grpc_end2end_test_config config);
+extern void retry_cancellation_pre_init(void);
+extern void retry_disabled(grpc_end2end_test_config config);
+extern void retry_disabled_pre_init(void);
+extern void retry_exceeds_buffer_size_in_initial_batch(grpc_end2end_test_config config);
+extern void retry_exceeds_buffer_size_in_initial_batch_pre_init(void);
+extern void retry_exceeds_buffer_size_in_subsequent_batch(grpc_end2end_test_config config);
+extern void retry_exceeds_buffer_size_in_subsequent_batch_pre_init(void);
+extern void retry_non_retriable_status(grpc_end2end_test_config config);
+extern void retry_non_retriable_status_pre_init(void);
+extern void retry_recv_initial_metadata(grpc_end2end_test_config config);
+extern void retry_recv_initial_metadata_pre_init(void);
+extern void retry_recv_message(grpc_end2end_test_config config);
+extern void retry_recv_message_pre_init(void);
+extern void retry_server_pushback_delay(grpc_end2end_test_config config);
+extern void retry_server_pushback_delay_pre_init(void);
+extern void retry_server_pushback_disabled(grpc_end2end_test_config config);
+extern void retry_server_pushback_disabled_pre_init(void);
+extern void retry_streaming(grpc_end2end_test_config config);
+extern void retry_streaming_pre_init(void);
+extern void retry_streaming_after_commit(grpc_end2end_test_config config);
+extern void retry_streaming_after_commit_pre_init(void);
+extern void retry_streaming_succeeds_before_replay_finished(grpc_end2end_test_config config);
+extern void retry_streaming_succeeds_before_replay_finished_pre_init(void);
+extern void retry_throttled(grpc_end2end_test_config config);
+extern void retry_throttled_pre_init(void);
+extern void retry_too_many_attempts(grpc_end2end_test_config config);
+extern void retry_too_many_attempts_pre_init(void);
 extern void server_finishes_request(grpc_end2end_test_config config);
 extern void server_finishes_request_pre_init(void);
 extern void shutdown_finishes_calls(grpc_end2end_test_config config);
@@ -200,6 +230,21 @@ void grpc_end2end_tests_pre_init(void) {
   request_with_flags_pre_init();
   request_with_payload_pre_init();
   resource_quota_server_pre_init();
+  retry_pre_init();
+  retry_cancellation_pre_init();
+  retry_disabled_pre_init();
+  retry_exceeds_buffer_size_in_initial_batch_pre_init();
+  retry_exceeds_buffer_size_in_subsequent_batch_pre_init();
+  retry_non_retriable_status_pre_init();
+  retry_recv_initial_metadata_pre_init();
+  retry_recv_message_pre_init();
+  retry_server_pushback_delay_pre_init();
+  retry_server_pushback_disabled_pre_init();
+  retry_streaming_pre_init();
+  retry_streaming_after_commit_pre_init();
+  retry_streaming_succeeds_before_replay_finished_pre_init();
+  retry_throttled_pre_init();
+  retry_too_many_attempts_pre_init();
   server_finishes_request_pre_init();
   shutdown_finishes_calls_pre_init();
   shutdown_finishes_tags_pre_init();
@@ -269,6 +314,21 @@ void grpc_end2end_tests(int argc, char **argv,
     request_with_flags(config);
     request_with_payload(config);
     resource_quota_server(config);
+    retry(config);
+    retry_cancellation(config);
+    retry_disabled(config);
+    retry_exceeds_buffer_size_in_initial_batch(config);
+    retry_exceeds_buffer_size_in_subsequent_batch(config);
+    retry_non_retriable_status(config);
+    retry_recv_initial_metadata(config);
+    retry_recv_message(config);
+    retry_server_pushback_delay(config);
+    retry_server_pushback_disabled(config);
+    retry_streaming(config);
+    retry_streaming_after_commit(config);
+    retry_streaming_succeeds_before_replay_finished(config);
+    retry_throttled(config);
+    retry_too_many_attempts(config);
     server_finishes_request(config);
     shutdown_finishes_calls(config);
     shutdown_finishes_tags(config);
@@ -468,6 +528,66 @@ void grpc_end2end_tests(int argc, char **argv,
       resource_quota_server(config);
       continue;
     }
+    if (0 == strcmp("retry", argv[i])) {
+      retry(config);
+      continue;
+    }
+    if (0 == strcmp("retry_cancellation", argv[i])) {
+      retry_cancellation(config);
+      continue;
+    }
+    if (0 == strcmp("retry_disabled", argv[i])) {
+      retry_disabled(config);
+      continue;
+    }
+    if (0 == strcmp("retry_exceeds_buffer_size_in_initial_batch", argv[i])) {
+      retry_exceeds_buffer_size_in_initial_batch(config);
+      continue;
+    }
+    if (0 == strcmp("retry_exceeds_buffer_size_in_subsequent_batch", argv[i])) {
+      retry_exceeds_buffer_size_in_subsequent_batch(config);
+      continue;
+    }
+    if (0 == strcmp("retry_non_retriable_status", argv[i])) {
+      retry_non_retriable_status(config);
+      continue;
+    }
+    if (0 == strcmp("retry_recv_initial_metadata", argv[i])) {
+      retry_recv_initial_metadata(config);
+      continue;
+    }
+    if (0 == strcmp("retry_recv_message", argv[i])) {
+      retry_recv_message(config);
+      continue;
+    }
+    if (0 == strcmp("retry_server_pushback_delay", argv[i])) {
+      retry_server_pushback_delay(config);
+      continue;
+    }
+    if (0 == strcmp("retry_server_pushback_disabled", argv[i])) {
+      retry_server_pushback_disabled(config);
+      continue;
+    }
+    if (0 == strcmp("retry_streaming", argv[i])) {
+      retry_streaming(config);
+      continue;
+    }
+    if (0 == strcmp("retry_streaming_after_commit", argv[i])) {
+      retry_streaming_after_commit(config);
+      continue;
+    }
+    if (0 == strcmp("retry_streaming_succeeds_before_replay_finished", argv[i])) {
+      retry_streaming_succeeds_before_replay_finished(config);
+      continue;
+    }
+    if (0 == strcmp("retry_throttled", argv[i])) {
+      retry_throttled(config);
+      continue;
+    }
+    if (0 == strcmp("retry_too_many_attempts", argv[i])) {
+      retry_too_many_attempts(config);
+      continue;
+    }
     if (0 == strcmp("server_finishes_request", argv[i])) {
       server_finishes_request(config);
       continue;
diff --git a/test/core/end2end/fixtures/h2_proxy.cc b/test/core/end2end/fixtures/h2_proxy.cc
index 93dde4f4a2..28a6eeeb78 100644
--- a/test/core/end2end/fixtures/h2_proxy.cc
+++ b/test/core/end2end/fixtures/h2_proxy.cc
@@ -49,7 +49,17 @@ static grpc_server* create_proxy_server(const char* port,
 
 static grpc_channel* create_proxy_client(const char* target,
                                          grpc_channel_args* client_args) {
-  return grpc_insecure_channel_create(target, client_args, nullptr);
+  // Disable retries in proxy client.
+  grpc_arg arg;
+  arg.type = GRPC_ARG_INTEGER;
+  arg.key = const_cast<char*>(GRPC_ARG_ENABLE_RETRIES);
+  arg.value.integer = 0;
+  grpc_channel_args* new_args =
+      grpc_channel_args_copy_and_add(client_args, &arg, 1);
+  grpc_channel* channel =
+      grpc_insecure_channel_create(target, new_args, nullptr);
+  grpc_channel_args_destroy(new_args);
+  return channel;
 }
 
 static const grpc_end2end_proxy_def proxy_def = {create_proxy_server,
diff --git a/test/core/end2end/fuzzers/hpack.dictionary b/test/core/end2end/fuzzers/hpack.dictionary
index 3ed82e19bd..c719d0fa84 100644
--- a/test/core/end2end/fuzzers/hpack.dictionary
+++ b/test/core/end2end/fuzzers/hpack.dictionary
@@ -21,7 +21,13 @@
 "\x0Auser-agent"
 "\x04host"
 "\x08lb-token"
+"\x1Agrpc-previous-rpc-attempts"
+"\x16grpc-retry-pushback-ms"
 "\x0Cgrpc-timeout"
+"\x011"
+"\x012"
+"\x013"
+"\x014"
 "\x00"
 "\x13grpc.wait_for_ready"
 "\x0Cgrpc.timeout"
@@ -32,8 +38,6 @@
 "\x04gzip"
 "\x0Bstream/gzip"
 "\x010"
-"\x011"
-"\x012"
 "\x08identity"
 "\x08trailers"
 "\x10application/grpc"
diff --git a/test/core/end2end/gen_build_yaml.py b/test/core/end2end/gen_build_yaml.py
index e7cf97b2d0..4e20b0b334 100755
--- a/test/core/end2end/gen_build_yaml.py
+++ b/test/core/end2end/gen_build_yaml.py
@@ -24,15 +24,24 @@ import hashlib
 
 FixtureOptions = collections.namedtuple(
     'FixtureOptions',
-    'fullstack includes_proxy dns_resolver name_resolution secure platforms ci_mac tracing exclude_configs exclude_iomgrs large_writes enables_compression supports_compression is_inproc is_http2 supports_proxy_auth supports_write_buffering')
+    'fullstack includes_proxy dns_resolver name_resolution secure platforms ci_mac tracing exclude_configs exclude_iomgrs large_writes enables_compression supports_compression is_inproc is_http2 supports_proxy_auth supports_write_buffering client_channel')
 default_unsecure_fixture_options = FixtureOptions(
-    True, False, True, True, False, ['windows', 'linux', 'mac', 'posix'], True, False, [], [], True, False, True, False, True, False, True)
-socketpair_unsecure_fixture_options = default_unsecure_fixture_options._replace(fullstack=False, dns_resolver=False)
-default_secure_fixture_options = default_unsecure_fixture_options._replace(secure=True)
-uds_fixture_options = default_unsecure_fixture_options._replace(dns_resolver=False, platforms=['linux', 'mac', 'posix'], exclude_iomgrs=['uv'])
+    True, False, True, True, False, ['windows', 'linux', 'mac', 'posix'],
+    True, False, [], [], True, False, True, False, True, False, True, True)
+socketpair_unsecure_fixture_options = default_unsecure_fixture_options._replace(
+    fullstack=False, dns_resolver=False, client_channel=False)
+default_secure_fixture_options = default_unsecure_fixture_options._replace(
+    secure=True)
+uds_fixture_options = default_unsecure_fixture_options._replace(
+    dns_resolver=False, platforms=['linux', 'mac', 'posix'],
+    exclude_iomgrs=['uv'])
 fd_unsecure_fixture_options = default_unsecure_fixture_options._replace(
-    dns_resolver=False, fullstack=False, platforms=['linux', 'mac', 'posix'], exclude_iomgrs=['uv'])
-inproc_fixture_options = default_unsecure_fixture_options._replace(dns_resolver=False, fullstack=False, name_resolution=False, supports_compression=False, is_inproc=True, is_http2=False, supports_write_buffering=False)
+    dns_resolver=False, fullstack=False, platforms=['linux', 'mac', 'posix'],
+    exclude_iomgrs=['uv'], client_channel=False)
+inproc_fixture_options = default_unsecure_fixture_options._replace(
+    dns_resolver=False, fullstack=False, name_resolution=False,
+    supports_compression=False, is_inproc=True, is_http2=False,
+    supports_write_buffering=False, client_channel=False)
 
 # maps fixture name to whether it requires the security library
 END2END_FIXTURES = {
@@ -68,9 +77,12 @@ END2END_FIXTURES = {
 
 TestOptions = collections.namedtuple(
     'TestOptions',
-    'needs_fullstack needs_dns needs_names proxyable secure traceable cpu_cost exclude_iomgrs large_writes flaky allows_compression needs_compression exclude_inproc needs_http2 needs_proxy_auth needs_write_buffering')
-default_test_options = TestOptions(False, False, False, True, False, True, 1.0, [], False, False, True, False, False, False, False, False)
-connectivity_test_options = default_test_options._replace(needs_fullstack=True)
+    'needs_fullstack needs_dns needs_names proxyable secure traceable cpu_cost exclude_iomgrs large_writes flaky allows_compression needs_compression exclude_inproc needs_http2 needs_proxy_auth needs_write_buffering needs_client_channel')
+default_test_options = TestOptions(
+    False, False, False, True, False, True, 1.0, [], False, False, True,
+    False, False, False, False, False, False)
+connectivity_test_options = default_test_options._replace(
+    needs_fullstack=True)
 
 LOWCPU = 0.1
 
@@ -80,9 +92,8 @@ END2END_TESTS = {
     'bad_hostname': default_test_options._replace(needs_names=True),
     'bad_ping': connectivity_test_options._replace(proxyable=False),
     'binary_metadata': default_test_options._replace(cpu_cost=LOWCPU),
-    'resource_quota_server': default_test_options._replace(large_writes=True,
-                                                           proxyable=False,
-                                                           allows_compression=False),
+    'resource_quota_server': default_test_options._replace(
+        large_writes=True, proxyable=False, allows_compression=False),
     'call_creds': default_test_options._replace(secure=True),
     'cancel_after_accept': default_test_options._replace(cpu_cost=LOWCPU),
     'cancel_after_client_done': default_test_options._replace(cpu_cost=LOWCPU),
@@ -91,18 +102,21 @@ END2END_TESTS = {
     'cancel_before_invoke': default_test_options._replace(cpu_cost=LOWCPU),
     'cancel_in_a_vacuum': default_test_options._replace(cpu_cost=LOWCPU),
     'cancel_with_status': default_test_options._replace(cpu_cost=LOWCPU),
-    'compressed_payload': default_test_options._replace(proxyable=False,needs_compression=True),
+    'compressed_payload': default_test_options._replace(proxyable=False,
+                                                        needs_compression=True),
     'connectivity': connectivity_test_options._replace(needs_names=True,
         proxyable=False, cpu_cost=LOWCPU, exclude_iomgrs=['uv']),
-    'default_host': default_test_options._replace(needs_fullstack=True,
-                                                  needs_dns=True,needs_names=True),
-    'disappearing_server': connectivity_test_options._replace(flaky=True,needs_names=True),
+    'default_host': default_test_options._replace(
+        needs_fullstack=True, needs_dns=True, needs_names=True),
+    'disappearing_server': connectivity_test_options._replace(flaky=True,
+                                                              needs_names=True),
     'empty_batch': default_test_options._replace(cpu_cost=LOWCPU),
     'filter_causes_close': default_test_options._replace(cpu_cost=LOWCPU),
     'filter_call_init_fails': default_test_options,
     'filter_latency': default_test_options._replace(cpu_cost=LOWCPU),
     'filter_status_code': default_test_options._replace(cpu_cost=LOWCPU),
-    'graceful_server_shutdown': default_test_options._replace(cpu_cost=LOWCPU,exclude_inproc=True),
+    'graceful_server_shutdown': default_test_options._replace(
+        cpu_cost=LOWCPU, exclude_inproc=True),
     'hpack_size': default_test_options._replace(proxyable=False,
                                                 traceable=False,
                                                 cpu_cost=LOWCPU),
@@ -127,30 +141,75 @@ END2END_TESTS = {
     'payload': default_test_options,
     'load_reporting_hook': default_test_options,
     'ping_pong_streaming': default_test_options._replace(cpu_cost=LOWCPU),
-    'ping': connectivity_test_options._replace(proxyable=False, cpu_cost=LOWCPU),
+    'ping': connectivity_test_options._replace(proxyable=False,
+                                               cpu_cost=LOWCPU),
     'proxy_auth': default_test_options._replace(needs_proxy_auth=True),
     'registered_call': default_test_options,
     'request_with_flags': default_test_options._replace(
         proxyable=False, cpu_cost=LOWCPU),
     'request_with_payload': default_test_options._replace(cpu_cost=LOWCPU),
+    # TODO(roth): Remove proxyable=False for all retry tests once we
+    # have a way for the proxy to propagate the fact that trailing
+    # metadata is available when initial metadata is returned.
+    # See https://github.com/grpc/grpc/issues/14467 for context.
+    'retry': default_test_options._replace(cpu_cost=LOWCPU,
+                                           needs_client_channel=True,
+                                           proxyable=False),
+    'retry_cancellation': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+    'retry_disabled': default_test_options._replace(cpu_cost=LOWCPU,
+                                                    needs_client_channel=True,
+                                                    proxyable=False),
+    'retry_exceeds_buffer_size_in_initial_batch': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+    'retry_exceeds_buffer_size_in_subsequent_batch':
+        default_test_options._replace(cpu_cost=LOWCPU,
+                                      needs_client_channel=True,
+                                      proxyable=False),
+    'retry_non_retriable_status': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+    'retry_recv_initial_metadata': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+    'retry_recv_message': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+    'retry_server_pushback_delay': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+    'retry_server_pushback_disabled': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+    'retry_streaming': default_test_options._replace(cpu_cost=LOWCPU,
+                                                     needs_client_channel=True,
+                                                     proxyable=False),
+    'retry_streaming_after_commit': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+    'retry_streaming_succeeds_before_replay_finished':
+        default_test_options._replace(cpu_cost=LOWCPU,
+                                      needs_client_channel=True,
+                                      proxyable=False),
+    'retry_throttled': default_test_options._replace(cpu_cost=LOWCPU,
+                                                     needs_client_channel=True,
+                                                     proxyable=False),
+    'retry_too_many_attempts': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
     'server_finishes_request': default_test_options._replace(cpu_cost=LOWCPU),
     'shutdown_finishes_calls': default_test_options._replace(cpu_cost=LOWCPU),
     'shutdown_finishes_tags': default_test_options._replace(cpu_cost=LOWCPU),
     'simple_cacheable_request': default_test_options._replace(cpu_cost=LOWCPU),
-    'stream_compression_compressed_payload': default_test_options._replace(proxyable=False,
-                                                               exclude_inproc=True),
-    'stream_compression_payload': default_test_options._replace(exclude_inproc=True),
-    'stream_compression_ping_pong_streaming': default_test_options._replace(exclude_inproc=True),
+    'stream_compression_compressed_payload': default_test_options._replace(
+        proxyable=False, exclude_inproc=True),
+    'stream_compression_payload': default_test_options._replace(
+        exclude_inproc=True),
+    'stream_compression_ping_pong_streaming': default_test_options._replace(
+        exclude_inproc=True),
     'simple_delayed_request': connectivity_test_options,
     'simple_metadata': default_test_options,
     'simple_request': default_test_options,
     'streaming_error_response': default_test_options._replace(cpu_cost=LOWCPU),
     'trailing_metadata': default_test_options,
     'workaround_cronet_compression': default_test_options,
-    'write_buffering': default_test_options._replace(cpu_cost=LOWCPU,
-                                                     needs_write_buffering=True),
-    'write_buffering_at_end': default_test_options._replace(cpu_cost=LOWCPU,
-                                                     needs_write_buffering=True),
+    'write_buffering': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_write_buffering=True),
+    'write_buffering_at_end': default_test_options._replace(
+        cpu_cost=LOWCPU, needs_write_buffering=True),
 }
 
 
@@ -191,6 +250,9 @@ def compatible(f, t):
   if END2END_TESTS[t].needs_write_buffering:
     if not END2END_FIXTURES[f].supports_write_buffering:
       return False
+  if END2END_TESTS[t].needs_client_channel:
+    if not END2END_FIXTURES[f].client_channel:
+      return False
   return True
 
 
diff --git a/test/core/end2end/generate_tests.bzl b/test/core/end2end/generate_tests.bzl
index bfd5d8525e..8e723fd609 100755
--- a/test/core/end2end/generate_tests.bzl
+++ b/test/core/end2end/generate_tests.bzl
@@ -24,7 +24,7 @@ def fixture_options(fullstack=True, includes_proxy=False, dns_resolver=True,
                     name_resolution=True, secure=True, tracing=False,
                     platforms=['windows', 'linux', 'mac', 'posix'],
                     is_inproc=False, is_http2=True, supports_proxy_auth=False,
-                    supports_write_buffering=True):
+                    supports_write_buffering=True, client_channel=True):
   return struct(
     fullstack=fullstack,
     includes_proxy=includes_proxy,
@@ -35,8 +35,9 @@ def fixture_options(fullstack=True, includes_proxy=False, dns_resolver=True,
     is_inproc=is_inproc,
     is_http2=is_http2,
     supports_proxy_auth=supports_proxy_auth,
-    supports_write_buffering=supports_write_buffering
-    #platforms=platforms
+    supports_write_buffering=supports_write_buffering,
+    client_channel=client_channel,
+    #platforms=platforms,
   )
 
 
@@ -47,6 +48,7 @@ END2END_FIXTURES = {
     'h2_load_reporting': fixture_options(),
     'h2_fakesec': fixture_options(),
     'h2_fd': fixture_options(dns_resolver=False, fullstack=False,
+                             client_channel=False,
                              platforms=['linux', 'mac', 'posix']),
     'h2_full': fixture_options(),
     'h2_full+pipe': fixture_options(platforms=['linux']),
@@ -55,24 +57,28 @@ END2END_FIXTURES = {
     'h2_http_proxy': fixture_options(supports_proxy_auth=True),
     'h2_oauth2': fixture_options(),
     'h2_proxy': fixture_options(includes_proxy=True),
-    'h2_sockpair_1byte': fixture_options(fullstack=False, dns_resolver=False),
-    'h2_sockpair': fixture_options(fullstack=False, dns_resolver=False),
+    'h2_sockpair_1byte': fixture_options(fullstack=False, dns_resolver=False,
+                                         client_channel=False),
+    'h2_sockpair': fixture_options(fullstack=False, dns_resolver=False,
+                                   client_channel=False),
     'h2_sockpair+trace': fixture_options(fullstack=False, dns_resolver=False,
-                                         tracing=True),
+                                         tracing=True, client_channel=False),
     'h2_ssl': fixture_options(secure=True),
     'h2_ssl_proxy': fixture_options(includes_proxy=True, secure=True),
     'h2_uds': fixture_options(dns_resolver=False,
                               platforms=['linux', 'mac', 'posix']),
     'inproc': fixture_options(fullstack=False, dns_resolver=False,
                               name_resolution=False, is_inproc=True,
-                              is_http2=False, supports_write_buffering=False),
+                              is_http2=False, supports_write_buffering=False,
+                              client_channel=False),
 }
 
 
 def test_options(needs_fullstack=False, needs_dns=False, needs_names=False,
                  proxyable=True, secure=False, traceable=False,
                  exclude_inproc=False, needs_http2=False,
-                 needs_proxy_auth=False, needs_write_buffering=False):
+                 needs_proxy_auth=False, needs_write_buffering=False,
+                 needs_client_channel=False):
   return struct(
     needs_fullstack=needs_fullstack,
     needs_dns=needs_dns,
@@ -83,7 +89,8 @@ def test_options(needs_fullstack=False, needs_dns=False, needs_names=False,
     exclude_inproc=exclude_inproc,
     needs_http2=needs_http2,
     needs_proxy_auth=needs_proxy_auth,
-    needs_write_buffering=needs_write_buffering
+    needs_write_buffering=needs_write_buffering,
+    needs_client_channel=needs_client_channel,
   )
 
 
@@ -118,7 +125,8 @@ END2END_TESTS = {
     'invoke_large_request': test_options(),
     'keepalive_timeout': test_options(proxyable=False, needs_http2=True),
     'large_metadata': test_options(),
-    'max_concurrent_streams': test_options(proxyable=False, exclude_inproc=True),
+    'max_concurrent_streams': test_options(proxyable=False,
+                                           exclude_inproc=True),
     'max_connection_age': test_options(exclude_inproc=True),
     'max_connection_idle': test_options(needs_fullstack=True, proxyable=False),
     'max_message_length': test_options(),
@@ -134,6 +142,37 @@ END2END_TESTS = {
     'registered_call': test_options(),
     'request_with_flags': test_options(proxyable=False),
     'request_with_payload': test_options(),
+    # TODO(roth): Remove proxyable=False for all retry tests once we
+    # have a way for the proxy to propagate the fact that trailing
+    # metadata is available when initial metadata is returned.
+    # See https://github.com/grpc/grpc/issues/14467 for context.
+    'retry': test_options(needs_client_channel=True, proxyable=False),
+    'retry_cancellation': test_options(needs_client_channel=True,
+                                       proxyable=False),
+    'retry_disabled': test_options(needs_client_channel=True, proxyable=False),
+    'retry_exceeds_buffer_size_in_initial_batch': test_options(
+        needs_client_channel=True, proxyable=False),
+    'retry_exceeds_buffer_size_in_subsequent_batch': test_options(
+        needs_client_channel=True, proxyable=False),
+    'retry_non_retriable_status': test_options(needs_client_channel=True,
+                                               proxyable=False),
+    'retry_recv_initial_metadata': test_options(needs_client_channel=True,
+                                                proxyable=False),
+    'retry_recv_message': test_options(needs_client_channel=True,
+                                       proxyable=False),
+    'retry_server_pushback_delay': test_options(needs_client_channel=True,
+                                                proxyable=False),
+    'retry_server_pushback_disabled': test_options(needs_client_channel=True,
+                                                   proxyable=False),
+    'retry_streaming': test_options(needs_client_channel=True, proxyable=False),
+    'retry_streaming_after_commit': test_options(needs_client_channel=True,
+                                                 proxyable=False),
+    'retry_streaming_succeeds_before_replay_finished': test_options(
+        needs_client_channel=True, proxyable=False),
+    'retry_throttled': test_options(needs_client_channel=True,
+                                    proxyable=False),
+    'retry_too_many_attempts': test_options(needs_client_channel=True,
+                                            proxyable=False),
     'server_finishes_request': test_options(),
     'shutdown_finishes_calls': test_options(),
     'shutdown_finishes_tags': test_options(),
@@ -142,7 +181,8 @@ END2END_TESTS = {
     'simple_metadata': test_options(),
     'simple_request': test_options(),
     'streaming_error_response': test_options(),
-    'stream_compression_compressed_payload': test_options(proxyable=False, exclude_inproc=True),
+    'stream_compression_compressed_payload': test_options(proxyable=False,
+                                                          exclude_inproc=True),
     'stream_compression_payload': test_options(exclude_inproc=True),
     'stream_compression_ping_pong_streaming': test_options(exclude_inproc=True),
     'trailing_metadata': test_options(),
@@ -183,6 +223,9 @@ def compatible(fopt, topt):
   if topt.needs_write_buffering:
     if not fopt.supports_write_buffering:
       return False
+  if topt.needs_client_channel:
+    if not fopt.client_channel:
+      return False
   return True
 
 
diff --git a/test/core/end2end/tests/retry.cc b/test/core/end2end/tests/retry.cc
new file mode 100644
index 0000000000..38ecc6fbb1
--- /dev/null
+++ b/test/core/end2end/tests/retry.cc
@@ -0,0 +1,325 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests a basic retry scenario:
+// - 2 retries allowed for ABORTED status
+// - first attempt returns ABORTED
+// - second attempt returns OK
+static void test_retry(grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 3,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  // Make sure the "grpc-previous-rpc-attempts" header was not sent in the
+  // initial attempt.
+  for (size_t i = 0; i < request_metadata_recv.count; ++i) {
+    GPR_ASSERT(!grpc_slice_eq(request_metadata_recv.metadata[i].key,
+                              GRPC_MDSTR_GRPC_PREVIOUS_RPC_ATTEMPTS));
+  }
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  grpc_call_unref(s);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_call_details_init(&call_details);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(201));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+  cq_verify(cqv);
+
+  // Make sure the "grpc-previous-rpc-attempts" header was sent in the retry.
+  bool found_retry_header = false;
+  for (size_t i = 0; i < request_metadata_recv.count; ++i) {
+    if (grpc_slice_eq(request_metadata_recv.metadata[i].key,
+                      GRPC_MDSTR_GRPC_PREVIOUS_RPC_ATTEMPTS)) {
+      GPR_ASSERT(
+          grpc_slice_eq(request_metadata_recv.metadata[i].value, GRPC_MDSTR_1));
+      found_retry_header = true;
+      break;
+    }
+  }
+  GPR_ASSERT(found_retry_header);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = response_payload;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_OK;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_OK);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 0);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry(config);
+}
+
+void retry_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_cancellation.cc b/test/core/end2end/tests/retry_cancellation.cc
new file mode 100644
index 0000000000..0504092c4f
--- /dev/null
+++ b/test/core/end2end/tests/retry_cancellation.cc
@@ -0,0 +1,277 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests retry cancellation.
+static void test_retry_cancellation(grpc_end2end_test_config config,
+                                    cancellation_mode mode) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 3,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    },\n"
+      "    \"timeout\": \"5s\"\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  char* name;
+  gpr_asprintf(&name, "retry_cancellation/%s", mode.name);
+  grpc_end2end_test_fixture f = begin_test(config, name, &client_args, nullptr);
+  gpr_free(name);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  // Client starts a batch with all 6 ops.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  // Server gets a call and fails with retryable status.
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  grpc_call_unref(s);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_call_details_init(&call_details);
+
+  // Server gets a second call (the retry).
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(201));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+  cq_verify(cqv);
+
+  // Initiate cancellation.
+  GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c, nullptr));
+
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == mode.expect_status);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_cancellation(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(cancellation_modes); ++i) {
+    test_retry_cancellation(config, cancellation_modes[i]);
+  }
+}
+
+void retry_cancellation_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_disabled.cc b/test/core/end2end/tests/retry_disabled.cc
new file mode 100644
index 0000000000..cb30502aeb
--- /dev/null
+++ b/test/core/end2end/tests/retry_disabled.cc
@@ -0,0 +1,262 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we don't retry when retries are disabled via the
+// GRPC_ARG_ENABLE_RETRIES channel arg, even when there is retry
+// configuration in the service config.
+// - 1 retry allowed for ABORTED status
+// - first attempt returns ABORTED but does not retry
+static void test_retry_disabled(grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg args[2];
+  args[0].type = GRPC_ARG_STRING;
+  args[0].key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  args[0].value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 2,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  args[1].type = GRPC_ARG_INTEGER;
+  args[1].key = const_cast<char*>(GRPC_ARG_ENABLE_RETRIES);
+  args[1].value.integer = 0;
+  grpc_channel_args client_args = {GPR_ARRAY_SIZE(args), args};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_disabled", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_disabled(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_disabled(config);
+}
+
+void retry_disabled_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_exceeds_buffer_size_in_initial_batch.cc b/test/core/end2end/tests/retry_exceeds_buffer_size_in_initial_batch.cc
new file mode 100644
index 0000000000..3908f29971
--- /dev/null
+++ b/test/core/end2end/tests/retry_exceeds_buffer_size_in_initial_batch.cc
@@ -0,0 +1,266 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we don't make any further attempts after we exceed the
+// max buffer size.
+// - 1 retry allowed for ABORTED status
+// - buffer size set to 2 bytes
+// - client sends a 3-byte message
+// - first attempt gets ABORTED but is not retried
+static void test_retry_exceeds_buffer_size_in_initial_batch(
+    grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg args[2];
+  args[0].type = GRPC_ARG_STRING;
+  args[0].key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  args[0].value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 2,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  args[1].type = GRPC_ARG_INTEGER;
+  args[1].key = const_cast<char*>(GRPC_ARG_PER_RPC_RETRY_BUFFER_SIZE);
+  args[1].value.integer = 2;
+  grpc_channel_args client_args = {GPR_ARRAY_SIZE(args), args};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_exceeds_buffer_size_in_initial_batch",
+                 &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_exceeds_buffer_size_in_initial_batch(
+    grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_exceeds_buffer_size_in_initial_batch(config);
+}
+
+void retry_exceeds_buffer_size_in_initial_batch_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_exceeds_buffer_size_in_subsequent_batch.cc b/test/core/end2end/tests/retry_exceeds_buffer_size_in_subsequent_batch.cc
new file mode 100644
index 0000000000..a8d7f0851a
--- /dev/null
+++ b/test/core/end2end/tests/retry_exceeds_buffer_size_in_subsequent_batch.cc
@@ -0,0 +1,276 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Similar to the retry_exceeds_buffer_size_in_initial_batch test, but we
+// don't exceed the buffer size until the second batch.
+// - 1 retry allowed for ABORTED status
+// - buffer size set to 100 KiB (larger than initial metadata)
+// - client sends a 100 KiB message
+// - first attempt gets ABORTED but is not retried
+static void test_retry_exceeds_buffer_size_in_subsequent_batch(
+    grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  char buf[102401];
+  memset(buf, 'a', sizeof(buf) - 1);
+  buf[sizeof(buf) - 1] = '\0';
+  grpc_slice request_payload_slice = grpc_slice_from_static_string(buf);
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg args[2];
+  args[0].type = GRPC_ARG_STRING;
+  args[0].key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  args[0].value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 2,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  args[1].type = GRPC_ARG_INTEGER;
+  args[1].key = const_cast<char*>(GRPC_ARG_PER_RPC_RETRY_BUFFER_SIZE);
+  args[1].value.integer = 102400;
+  grpc_channel_args client_args = {GPR_ARRAY_SIZE(args), args};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_exceeds_buffer_size_in_subsequent_batch",
+                 &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(2), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_exceeds_buffer_size_in_subsequent_batch(
+    grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_exceeds_buffer_size_in_subsequent_batch(config);
+}
+
+void retry_exceeds_buffer_size_in_subsequent_batch_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_non_retriable_status.cc b/test/core/end2end/tests/retry_non_retriable_status.cc
new file mode 100644
index 0000000000..6d59db0b04
--- /dev/null
+++ b/test/core/end2end/tests/retry_non_retriable_status.cc
@@ -0,0 +1,257 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we don't retry for non-retryable status codes.
+// - 1 retry allowed for ABORTED status
+// - first attempt gets INVALID_ARGUMENT, so no retry is done
+static void test_retry_non_retriable_status(grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 2,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_non_retriable_status", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_INVALID_ARGUMENT;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_INVALID_ARGUMENT);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_non_retriable_status(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_non_retriable_status(config);
+}
+
+void retry_non_retriable_status_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_recv_initial_metadata.cc b/test/core/end2end/tests/retry_recv_initial_metadata.cc
new file mode 100644
index 0000000000..14215e449c
--- /dev/null
+++ b/test/core/end2end/tests/retry_recv_initial_metadata.cc
@@ -0,0 +1,268 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that receiving initial metadata commits the call.
+// - 1 retry allowed for ABORTED status
+// - first attempt receives initial metadata before trailing metadata,
+//   so no retry is done even though status was ABORTED
+static void test_retry_recv_initial_metadata(grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 2,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_recv_initial_metadata", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  // Server sends initial metadata in its own batch, before sending
+  // trailing metadata.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_recv_initial_metadata(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_recv_initial_metadata(config);
+}
+
+void retry_recv_initial_metadata_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_recv_message.cc b/test/core/end2end/tests/retry_recv_message.cc
new file mode 100644
index 0000000000..86171fdc01
--- /dev/null
+++ b/test/core/end2end/tests/retry_recv_message.cc
@@ -0,0 +1,261 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that receiving a message commits the call.
+// - 1 retry allowed for ABORTED status
+// - first attempt receives a message and therefore does not retry even
+//   though the final status is ABORTED
+static void test_retry_recv_message(grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 2,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_recv_message", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = response_payload;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_recv_message(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_recv_message(config);
+}
+
+void retry_recv_message_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_server_pushback_delay.cc b/test/core/end2end/tests/retry_server_pushback_delay.cc
new file mode 100644
index 0000000000..1da986041f
--- /dev/null
+++ b/test/core/end2end/tests/retry_server_pushback_delay.cc
@@ -0,0 +1,318 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we honor server push-back delay.
+// - 2 retries allowed for ABORTED status
+// - first attempt gets ABORTED with a long delay
+// - second attempt succeeds
+static void test_retry_server_pushback_delay(grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_metadata pushback_md;
+  memset(&pushback_md, 0, sizeof(pushback_md));
+  pushback_md.key = GRPC_MDSTR_GRPC_RETRY_PUSHBACK_MS;
+  pushback_md.value = grpc_slice_from_static_string("2000");
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 3,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_server_pushback_delay", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 1;
+  op->data.send_status_from_server.trailing_metadata = &pushback_md;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  gpr_timespec before_retry = gpr_now(GPR_CLOCK_MONOTONIC);
+
+  grpc_call_unref(s);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_call_details_init(&call_details);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(201));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+  cq_verify(cqv);
+
+  gpr_timespec after_retry = gpr_now(GPR_CLOCK_MONOTONIC);
+  gpr_timespec retry_delay = gpr_time_sub(after_retry, before_retry);
+  // Configured back-off was 1 second, server push-back said 2 seconds.
+  // To avoid flakiness, we allow some fudge factor here.
+  gpr_log(GPR_INFO, "retry delay was {.tv_sec=%" PRId64 ", .tv_nsec=%d}",
+          retry_delay.tv_sec, retry_delay.tv_nsec);
+  GPR_ASSERT(retry_delay.tv_sec >= 1);
+  if (retry_delay.tv_sec == 1) {
+    GPR_ASSERT(retry_delay.tv_nsec >= 900000000);
+  }
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_OK;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_OK);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 0);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_server_pushback_delay(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_server_pushback_delay(config);
+}
+
+void retry_server_pushback_delay_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_server_pushback_disabled.cc b/test/core/end2end/tests/retry_server_pushback_disabled.cc
new file mode 100644
index 0000000000..13d4f01eea
--- /dev/null
+++ b/test/core/end2end/tests/retry_server_pushback_disabled.cc
@@ -0,0 +1,306 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we don't retry when disabled by server push-back.
+// - 2 retries allowed for ABORTED status
+// - first attempt gets ABORTED
+// - second attempt gets ABORTED but server push back disables retrying
+static void test_retry_server_pushback_disabled(
+    grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_metadata pushback_md;
+  memset(&pushback_md, 0, sizeof(pushback_md));
+  pushback_md.key = GRPC_MDSTR_GRPC_RETRY_PUSHBACK_MS;
+  pushback_md.value = grpc_slice_from_static_string("-1");
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 3,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f = begin_test(
+      config, "retry_server_pushback_disabled", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  grpc_call_unref(s);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_call_details_init(&call_details);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(201));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 1;
+  op->data.send_status_from_server.trailing_metadata = &pushback_md;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_server_pushback_disabled(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_server_pushback_disabled(config);
+}
+
+void retry_server_pushback_disabled_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_streaming.cc b/test/core/end2end/tests/retry_streaming.cc
new file mode 100644
index 0000000000..e96e57e8bc
--- /dev/null
+++ b/test/core/end2end/tests/retry_streaming.cc
@@ -0,0 +1,424 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests retrying a streaming RPC.  This is the same as
+// the basic retry test, except that the client sends two messages on the
+// call before the initial attempt fails.
+// FIXME: We should also test the case where the retry is committed after
+// replaying 1 of 2 previously-completed send_message ops.  However,
+// there's no way to trigger that from an end2end test, because the
+// replayed ops happen under the hood -- they are not surfaced to the
+// C-core API, and therefore we have no way to inject the commit at the
+// right point.
+static void test_retry_streaming(grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice request2_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_slice request3_payload_slice = grpc_slice_from_static_string("baz");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("quux");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* request2_payload =
+      grpc_raw_byte_buffer_create(&request2_payload_slice, 1);
+  grpc_byte_buffer* request3_payload =
+      grpc_raw_byte_buffer_create(&request3_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* request2_payload_recv = nullptr;
+  grpc_byte_buffer* request3_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 3,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_streaming", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  // Client starts a batch for receiving initial metadata, a message,
+  // and trailing metadata.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  // Client sends initial metadata and a message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(2), true);
+  cq_verify(cqv);
+
+  // Server gets a call with received initial metadata.
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  // Server receives a message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  // Client sends a second message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request2_payload;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(3), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(3), true);
+  cq_verify(cqv);
+
+  // Server receives the second message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request2_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+  cq_verify(cqv);
+
+  // Server sends both initial and trailing metadata.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(104), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(104), true);
+  cq_verify(cqv);
+
+  // Clean up from first attempt.
+  grpc_call_unref(s);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_call_details_init(&call_details);
+  GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+  grpc_byte_buffer_destroy(request_payload_recv);
+  request_payload_recv = nullptr;
+  GPR_ASSERT(
+      byte_buffer_eq_slice(request2_payload_recv, request2_payload_slice));
+  grpc_byte_buffer_destroy(request2_payload_recv);
+  request2_payload_recv = nullptr;
+
+  // Server gets a second call (the retry).
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(201));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  // Server receives a message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+  cq_verify(cqv);
+
+  // Server receives a second message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request2_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(203), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(203), true);
+  cq_verify(cqv);
+
+  // Client sends a third message and a close.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request3_payload;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(4), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(4), true);
+  cq_verify(cqv);
+
+  // Server receives a third message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request3_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(204), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(204), true);
+  cq_verify(cqv);
+
+  // Server receives a close and sends initial metadata, a message, and
+  // trailing metadata.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = response_payload;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  // Returning a retriable code, but because we are also sending a
+  // message, the client will commit instead of retrying again.
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(205), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(205), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(request2_payload);
+  grpc_byte_buffer_destroy(request3_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+  grpc_byte_buffer_destroy(request_payload_recv);
+  GPR_ASSERT(
+      byte_buffer_eq_slice(request2_payload_recv, request2_payload_slice));
+  grpc_byte_buffer_destroy(request2_payload_recv);
+  GPR_ASSERT(
+      byte_buffer_eq_slice(request3_payload_recv, request3_payload_slice));
+  grpc_byte_buffer_destroy(request3_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_streaming(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_streaming(config);
+}
+
+void retry_streaming_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_streaming_after_commit.cc b/test/core/end2end/tests/retry_streaming_after_commit.cc
new file mode 100644
index 0000000000..43eee86d95
--- /dev/null
+++ b/test/core/end2end/tests/retry_streaming_after_commit.cc
@@ -0,0 +1,354 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we can continue to send/recv messages on a streaming call
+// after retries are committed.
+static void test_retry_streaming_after_commit(grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice request2_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("baz");
+  grpc_slice response2_payload_slice = grpc_slice_from_static_string("quux");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* request2_payload =
+      grpc_raw_byte_buffer_create(&request2_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* response2_payload =
+      grpc_raw_byte_buffer_create(&response2_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* request2_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_byte_buffer* response2_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 3,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_streaming_after_commit", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  // Client starts a batch for receiving initial metadata and a message.
+  // This will commit retries.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  // Client sends initial metadata and a message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(3), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(3), true);
+  cq_verify(cqv);
+
+  // Server gets a call with received initial metadata.
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  // Server receives a message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  // Server sends initial metadata and a message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = response_payload;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+  // Client receives initial metadata and a message.
+  CQ_EXPECT_COMPLETION(cqv, tag(2), true);
+  cq_verify(cqv);
+
+  // Client sends a second message and a close.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request2_payload;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(4), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(4), true);
+  cq_verify(cqv);
+
+  // Server receives a second message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request2_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(104), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(104), true);
+  cq_verify(cqv);
+
+  // Server receives a close, sends a second message, and sends status.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = response2_payload;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  // Returning a retriable code, but because retries are already
+  // committed, the client will not retry.
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(105), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(105), true);
+  cq_verify(cqv);
+
+  // Client receives a second message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response2_payload_recv;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(5), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(5), true);
+  cq_verify(cqv);
+
+  // Client receives status.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(request2_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(response2_payload);
+  GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+  grpc_byte_buffer_destroy(request_payload_recv);
+  GPR_ASSERT(
+      byte_buffer_eq_slice(request2_payload_recv, request2_payload_slice));
+  grpc_byte_buffer_destroy(request2_payload_recv);
+  GPR_ASSERT(
+      byte_buffer_eq_slice(response_payload_recv, response_payload_slice));
+  grpc_byte_buffer_destroy(response_payload_recv);
+  GPR_ASSERT(
+      byte_buffer_eq_slice(response2_payload_recv, response2_payload_slice));
+  grpc_byte_buffer_destroy(response2_payload_recv);
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_streaming_after_commit(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_streaming_after_commit(config);
+}
+
+void retry_streaming_after_commit_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_streaming_succeeds_before_replay_finished.cc b/test/core/end2end/tests/retry_streaming_succeeds_before_replay_finished.cc
new file mode 100644
index 0000000000..5c92f64805
--- /dev/null
+++ b/test/core/end2end/tests/retry_streaming_succeeds_before_replay_finished.cc
@@ -0,0 +1,400 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we correctly clean up if the second attempt finishes
+// before we have finished replaying all of the send ops.
+static void test_retry_streaming_succeeds_before_replay_finished(
+    grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice request2_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_slice request3_payload_slice = grpc_slice_from_static_string("baz");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("quux");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* request2_payload =
+      grpc_raw_byte_buffer_create(&request2_payload_slice, 1);
+  grpc_byte_buffer* request3_payload =
+      grpc_raw_byte_buffer_create(&request3_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* request2_payload_recv = nullptr;
+  grpc_byte_buffer* request3_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 3,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_streaming", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  // Client starts a batch for receiving initial metadata, a message,
+  // and trailing metadata.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  // Client sends initial metadata and a message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(2), true);
+  cq_verify(cqv);
+
+  // Server gets a call with received initial metadata.
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  // Server receives a message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  // Client sends a second message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request2_payload;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(3), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(3), true);
+  cq_verify(cqv);
+
+  // Server receives the second message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request2_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+  cq_verify(cqv);
+
+  // Client sends a third message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request3_payload;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(4), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(4), true);
+  cq_verify(cqv);
+
+  // Server receives the third message.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request3_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(104), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(104), true);
+  cq_verify(cqv);
+
+  // Server sends both initial and trailing metadata.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(105), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(105), true);
+  cq_verify(cqv);
+
+  // Clean up from first attempt.
+  grpc_call_unref(s);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_call_details_init(&call_details);
+  GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+  grpc_byte_buffer_destroy(request_payload_recv);
+  request_payload_recv = nullptr;
+  GPR_ASSERT(
+      byte_buffer_eq_slice(request2_payload_recv, request2_payload_slice));
+  grpc_byte_buffer_destroy(request2_payload_recv);
+  request2_payload_recv = nullptr;
+  GPR_ASSERT(
+      byte_buffer_eq_slice(request3_payload_recv, request3_payload_slice));
+  grpc_byte_buffer_destroy(request3_payload_recv);
+  request3_payload_recv = nullptr;
+
+  // Server gets a second call (the retry).
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(201));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  // Server receives the first message (and does not receive any others).
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request_payload_recv;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+  cq_verify(cqv);
+
+  // Server sends initial metadata, a message, and trailing metadata.
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = response_payload;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  // Returning a retriable code, but because we are also sending a
+  // message, the client will commit instead of retrying again.
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(205), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(205), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(request2_payload);
+  grpc_byte_buffer_destroy(request3_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_streaming_succeeds_before_replay_finished(
+    grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_streaming_succeeds_before_replay_finished(config);
+}
+
+void retry_streaming_succeeds_before_replay_finished_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_throttled.cc b/test/core/end2end/tests/retry_throttled.cc
new file mode 100644
index 0000000000..8cd08487ea
--- /dev/null
+++ b/test/core/end2end/tests/retry_throttled.cc
@@ -0,0 +1,264 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we don't retry when throttled.
+// - 1 retry allowed for ABORTED status
+// - first attempt gets ABORTED but is over limit, so no retry is done
+static void test_retry_throttled(grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 2,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ],\n"
+      // A single failure will cause us to be throttled.
+      // (This is not a very realistic config, but it works for the
+      // purposes of this test.)
+      "  \"retryThrottling\": {\n"
+      "    \"maxTokens\": 2,\n"
+      "    \"tokenRatio\": 1.0,\n"
+      "  }\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_throttled", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_throttled(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_throttled(config);
+}
+
+void retry_throttled_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_too_many_attempts.cc b/test/core/end2end/tests/retry_too_many_attempts.cc
new file mode 100644
index 0000000000..5225c9b229
--- /dev/null
+++ b/test/core/end2end/tests/retry_too_many_attempts.cc
@@ -0,0 +1,299 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+  return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         nullptr)
+                 .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we stop retrying after the configured number of attempts.
+// - 1 retry allowed for ABORTED status
+// - first attempt gets ABORTED
+// - second attempt gets ABORTED but does not retry
+static void test_retry_too_many_attempts(grpc_end2end_test_config config) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+  grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+
+  grpc_arg arg;
+  arg.type = GRPC_ARG_STRING;
+  arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+  arg.value.string = const_cast<char*>(
+      "{\n"
+      "  \"methodConfig\": [ {\n"
+      "    \"name\": [\n"
+      "      { \"service\": \"service\", \"method\": \"method\" }\n"
+      "    ],\n"
+      "    \"retryPolicy\": {\n"
+      "      \"maxAttempts\": 2,\n"
+      "      \"initialBackoff\": \"1s\",\n"
+      "      \"maxBackoff\": \"120s\",\n"
+      "      \"backoffMultiplier\": 1.6,\n"
+      "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+      "    }\n"
+      "  } ]\n"
+      "}");
+  grpc_channel_args client_args = {1, &arg};
+  grpc_end2end_test_fixture f =
+      begin_test(config, "retry_too_many_attempts", &client_args, nullptr);
+
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+
+  gpr_timespec deadline = five_seconds_from_now();
+  c = grpc_channel_create_call(
+      f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+      grpc_slice_from_static_string("/service/method"),
+      get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+      nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  grpc_call_unref(s);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_call_details_init(&call_details);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(201));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+  cq_verify(cqv);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+  op->data.send_status_from_server.status_details = &status_details;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void retry_too_many_attempts(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+  test_retry_too_many_attempts(config);
+}
+
+void retry_too_many_attempts_pre_init(void) {}
diff --git a/test/core/transport/BUILD b/test/core/transport/BUILD
index b31d4ff899..2c2d05b9ae 100644
--- a/test/core/transport/BUILD
+++ b/test/core/transport/BUILD
@@ -119,3 +119,15 @@ grpc_cc_test(
         "//test/core/util:grpc_test_util",
     ],
 )
+
+grpc_cc_test(
+    name = "status_metadata_test",
+    srcs = ["status_metadata_test.cc"],
+    language = "C++",
+    deps = [
+        "//:grpc",
+    ],
+    external_deps = [
+        "gtest",
+    ],
+)
diff --git a/test/core/transport/status_metadata_test.cc b/test/core/transport/status_metadata_test.cc
new file mode 100644
index 0000000000..a96f11c1ea
--- /dev/null
+++ b/test/core/transport/status_metadata_test.cc
@@ -0,0 +1,61 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "src/core/lib/transport/status_metadata.h"
+
+#include <gtest/gtest.h>
+
+#include "src/core/lib/transport/static_metadata.h"
+
+namespace {
+
+TEST(GetStatusCodeFromMetadata, OK) {
+  EXPECT_EQ(GRPC_STATUS_OK,
+            grpc_get_status_code_from_metadata(GRPC_MDELEM_GRPC_STATUS_0));
+}
+
+TEST(GetStatusCodeFromMetadata, CANCELLED) {
+  EXPECT_EQ(GRPC_STATUS_CANCELLED,
+            grpc_get_status_code_from_metadata(GRPC_MDELEM_GRPC_STATUS_1));
+}
+
+TEST(GetStatusCodeFromMetadata, UNKNOWN) {
+  EXPECT_EQ(GRPC_STATUS_UNKNOWN,
+            grpc_get_status_code_from_metadata(GRPC_MDELEM_GRPC_STATUS_2));
+}
+
+TEST(GetStatusCodeFromMetadata, Other) {
+  grpc_mdelem status_md = grpc_mdelem_from_slices(
+      GRPC_MDSTR_GRPC_STATUS, grpc_slice_from_static_string("10"));
+  EXPECT_EQ(GRPC_STATUS_ABORTED, grpc_get_status_code_from_metadata(status_md));
+  GRPC_MDELEM_UNREF(status_md);
+}
+
+TEST(GetStatusCodeFromMetadata, Unparseable) {
+  grpc_mdelem status_md = grpc_mdelem_from_slices(
+      GRPC_MDSTR_GRPC_STATUS, grpc_slice_from_static_string("NaN"));
+  EXPECT_EQ(GRPC_STATUS_UNKNOWN, grpc_get_status_code_from_metadata(status_md));
+  GRPC_MDELEM_UNREF(status_md);
+}
+
+}  // namespace
+
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}