Don't retry on SubscribeToShard and checkstyle fixes

Author: shorea

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/NettyNioAsyncHttpClient.java

@@ -148,13 +148,12 @@ protected ChannelPool newPool(URI key) {
                         .option(ChannelOption.TCP_NODELAY, true)
                         .remoteAddress(key.getHost(), key.getPort());
                 AtomicReference<ChannelPool> channelPoolRef = new AtomicReference<>();
-                channelPoolRef.set(
-                    new ReleaseOnceChannelPool(
-                        new HandlerRemovingChannelPool(
-                            new HttpOrHttp2ChannelPool(bootstrap,
-                                                       new ChannelPipelineInitializer(protocol, sslContext, maxStreams, channelPoolRef),
-                                                       configuration.maxConnections(),
-                                                       configuration))));
+                ChannelPipelineInitializer handler =
+                    new ChannelPipelineInitializer(protocol, sslContext, maxStreams, channelPoolRef);
+                channelPoolRef.set(new ReleaseOnceChannelPool(
+                    new HandlerRemovingChannelPool(
+                        new HttpOrHttp2ChannelPool(bootstrap, handler,
+                                                   configuration.maxConnections(), configuration))));
                 return channelPoolRef.get();
             }
         };

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ReleaseOnceChannelPool.java

@@ -1,11 +1,11 @@
 /*
- * Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * Copyright 2010-2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.
  * A copy of the License is located at
  *
- *     http://www.apache.org/licenses/LICENSE-2.0
+ *  http://aws.amazon.com/apache2.0
  *
  * or in the "license" file accompanying this file. This file is distributed
  * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
@@ -24,13 +24,15 @@
 import io.netty.util.concurrent.Promise;
 import io.netty.util.concurrent.SucceededFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
+import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.http.nio.netty.internal.http2.Http2MultiplexedChannelPool;
 
 /**
  * Wrapper around a {@link ChannelPool} to protect it from having the same channel released twice. This can
  * cause issues in {@link FixedChannelPool} and {@link Http2MultiplexedChannelPool} which has a simple
  * mechanism to track leased connections.
  */
+@SdkInternalApi
 public class ReleaseOnceChannelPool implements ChannelPool {
 
     private static final AttributeKey<AtomicBoolean> IS_RELEASED = AttributeKey.newInstance("isReleased");
@@ -53,7 +55,7 @@ public Future<Channel> acquire(Promise<Channel> promise) {
 
     private GenericFutureListener<Future<Channel>> onAcquire() {
         return future -> {
-            if(future.isSuccess()) {
+            if (future.isSuccess()) {
                 future.getNow().attr(IS_RELEASED).set(new AtomicBoolean(false));
             }
         };

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ResponseHandler.java

@@ -51,7 +51,6 @@
 import software.amazon.awssdk.http.SdkHttpResponse;
 import software.amazon.awssdk.http.nio.netty.internal.http2.Http2ResetSendingSubscription;
 import software.amazon.awssdk.utils.FunctionalUtils.UnsafeRunnable;
-import software.amazon.awssdk.utils.async.DelegatingSubscriber;
 import software.amazon.awssdk.utils.async.DelegatingSubscription;
 
 @Sharable
@@ -211,12 +210,13 @@ private Subscription resolveSubscription(Subscription subscription) {
                 private void onCancel() {
                     try {
                         isCancelled.set(true);
-                        requestContext.handler().exceptionOccurred(new RuntimeException("Cancelled subscription"));
+                        requestContext.handler().exceptionOccurred(
+                            new RuntimeException("Subscriber cancelled before all events were published"));
                     } finally {
                         runAndLogError("Could not release channel back to the pool",
-                                       () -> closeAndRelease(channelContext));
+                            () -> closeAndRelease(channelContext));
                         runAndLogError("Could not release channel back to the pool",
-                                       () -> closeAndRelease(channelContext));
+                            () -> closeAndRelease(channelContext));
                     }
                 }
 

services/kinesis/src/it/java/software/amazon/awssdk/services/kinesis/SubscribeToShardIntegrationTest.java

@@ -121,7 +121,7 @@ public void complete() {
                                     }).join();
             fail("Expected exception");
         } catch (CompletionException e) {
-            assertThat(e.getCause()).hasMessageContaining("cancelled");
+            assertThat(e.getCause().getCause()).hasMessageContaining("cancelled");
             assertThat(terminalCalled).as("complete or onComplete was called when it shouldn't have been")
                                       .isFalse();
         }

services/kinesis/src/main/java/software/amazon/awssdk/services/kinesis/KinesisRetryPolicy.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2010-2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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.
+ */
+
+package software.amazon.awssdk.services.kinesis;
+
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.awscore.retry.AwsRetryPolicy;
+import software.amazon.awssdk.core.retry.RetryPolicy;
+import software.amazon.awssdk.core.retry.conditions.AndRetryCondition;
+import software.amazon.awssdk.services.kinesis.model.SubscribeToShardRequest;
+
+/**
+ * Default retry policy for the Kinesis Client.
+ */
+@SdkInternalApi
+public class KinesisRetryPolicy {
+
+    /**
+     * Default retry policy for Kinesis. Turns off retries for SubscribeToShard
+     */
+    private static final RetryPolicy DEFAULT =
+        AwsRetryPolicy.defaultRetryPolicy().toBuilder()
+                      .retryCondition(AndRetryCondition.create(
+                          c -> !(c.originalRequest() instanceof SubscribeToShardRequest),
+                          AwsRetryPolicy.defaultRetryCondition()))
+                      .build();
+
+    private KinesisRetryPolicy() {
+
+    }
+
+    /**
+     * @return Default retry policy used by Kinesis
+     */
+    public static RetryPolicy defaultPolicy() {
+        return DEFAULT;
+    }
+}
+

services/kinesis/src/main/resources/codegen-resources/customization.config

@@ -6,5 +6,6 @@
         "protocol": "cbor"
     },
     "skipSmokeTests": "true",
-    "blacklistedSimpleMethods": ["deregisterStreamConsumer", "describeStreamConsumer", "listShards"]
+    "blacklistedSimpleMethods": ["deregisterStreamConsumer", "describeStreamConsumer", "listShards"],
+     "customRetryPolicy" : "software.amazon.awssdk.services.kinesis.KinesisRetryPolicy"
 }