Skip to content

Commit 3c2f67f

Browse files
shoreashorea
committed
Syncing client with latest model
1 parent a88fd4b commit 3c2f67f

File tree

12 files changed

+2249
-629
lines changed

12 files changed

+2249
-629
lines changed

‎bundle/pom.xml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,7 @@
7474
<include>commons-codec:commons-codec</include>
7575
<include>software.amazon.awssdk.ion:ion-java</include>
7676
<include>software.amazon.awssdk:*</include>
77+
<include>software.amazon:*</include>
7778
<include>commons-logging:*</include>
7879
</includes>
7980
</artifactSet>

‎core/src/main/java/software/amazon/awssdk/core/auth/Aws4Signer.java

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@
3030
import java.util.concurrent.TimeUnit;
3131
import org.slf4j.Logger;
3232
import org.slf4j.LoggerFactory;
33-
import software.amazon.awssdk.annotations.ReviewBeforeRelease;
3433
import software.amazon.awssdk.annotations.SdkTestInternalApi;
3534
import software.amazon.awssdk.core.auth.internal.Aws4SignerRequestParams;
3635
import software.amazon.awssdk.core.auth.internal.Aws4SignerUtils;
@@ -56,10 +55,7 @@ public class Aws4Signer extends AbstractAwsSigner
5655

5756
private static final int SIGNER_CACHE_MAX_SIZE = 300;
5857
private static final FifoCache<SignerKey> SIGNER_CACHE = new FifoCache<>(SIGNER_CACHE_MAX_SIZE);
59-
// TODO sigv4 auth
60-
@ReviewBeforeRelease("Remove this when Sigv4 auth is implemented properly service side.")
61-
private static final List<String> LIST_OF_HEADERS_TO_IGNORE_IN_LOWER_CASE = Arrays.asList("connection", "x-amzn-trace-id", "host");
62-
//private static final List<String> LIST_OF_HEADERS_TO_IGNORE_IN_LOWER_CASE = Arrays.asList("connection", "x-amzn-trace-id");
58+
private static final List<String> LIST_OF_HEADERS_TO_IGNORE_IN_LOWER_CASE = Arrays.asList("connection", "x-amzn-trace-id" );
6359

6460
/**
6561
* Service name override for use when the endpoint can't be used to

‎core/src/main/java/software/amazon/awssdk/core/http/pipeline/stages/ApplyTransactionIdStage.java

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -39,9 +39,6 @@ public class ApplyTransactionIdStage implements MutableRequestToRequestPipeline
3939
public SdkHttpFullRequest.Builder execute(SdkHttpFullRequest.Builder request, RequestExecutionContext context)
4040
throws Exception {
4141
InterruptMonitor.checkInterrupted();
42-
// for (int i = 1; i < 50; i++) {
43-
// request.header("x-amz-shorea-" + i, UUID.randomUUID().toString());
44-
// }
4542
return request.header(HEADER_SDK_TRANSACTION_ID, new UUID(random.nextLong(), random.nextLong()).toString());
4643
}
4744
}

‎flow/src/main/java/software/amazon/eventstream/MessageDecoder.java

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,18 +15,23 @@ public final class MessageDecoder {
1515

1616
public MessageDecoder(Consumer<Message> messageConsumer) {
1717
this.messageConsumer = messageConsumer;
18-
this.buf = ByteBuffer.allocate(128 * 1024);
18+
this.buf = ByteBuffer.allocate(16 * 128 * 1024);
1919
}
2020

2121
public void feed(byte[] bytes) {
2222
feed(bytes, 0, bytes.length);
2323
}
2424

2525
public void feed(byte[] bytes, int offset, int length) {
26+
// TODO instead of blindly putting bytes can we first calculate the prelude if available then cache, then
27+
// put all bytes up to the length of the frame then decode then clear and reset for next prelude
28+
// Also we can detect the length of the payload and grow if needed
2629
buf.put(bytes, offset, length);
2730
ByteBuffer readView = (ByteBuffer) buf.duplicate().flip();
2831
int bytesConsumed = 0;
2932
while (readView.remaining() >= 15) {
33+
// TODO seems wasteful to decode the prelude multiple times. Can we cache until we've read all the data for
34+
// this frame?
3035
int totalMessageLength = toIntExact(Prelude.decode(readView.duplicate()).getTotalLength());
3136

3237
if (readView.remaining() >= totalMessageLength) {

‎services/kinesis/pom.xml

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -31,10 +31,14 @@
3131
<url>https://aws.amazon.com/sdkforjava</url>
3232

3333
<dependencies>
34-
<dependency>
35-
<groupId>log4j</groupId>
36-
<artifactId>log4j</artifactId>
37-
</dependency>
34+
<!--<dependency>-->
35+
<!--<groupId>log4j</groupId>-->
36+
<!--<artifactId>log4j</artifactId>-->
37+
<!--</dependency>-->
38+
<!--<dependency>-->
39+
<!--<groupId>org.slf4j</groupId>-->
40+
<!--<artifactId>slf4j-log4j12</artifactId>-->
41+
<!--</dependency>-->
3842
<dependency>
3943
<groupId>software.amazon.awssdk</groupId>
4044
<artifactId>netty-nio-client</artifactId>

‎services/kinesis/src/it/java/log4j.properties

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
log4j.rootLogger=INFO, A1
1+
log4j.rootLogger=DEBUG, A1
22
log4j.appender.A1=org.apache.log4j.ConsoleAppender
33
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
44

‎services/kinesis/src/it/java/software/amazon/awssdk/services/kinesis/H2Demo.java

Lines changed: 91 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,10 @@
66
import java.util.Random;
77
import java.util.UUID;
88
import java.util.concurrent.CompletableFuture;
9+
import java.util.concurrent.CountDownLatch;
910
import java.util.concurrent.ExecutorService;
1011
import java.util.concurrent.Executors;
12+
import java.util.concurrent.TimeUnit;
1113
import java.util.concurrent.atomic.AtomicInteger;
1214
import org.apache.log4j.BasicConfigurator;
1315
import org.reactivestreams.Subscriber;
@@ -20,78 +22,123 @@
2022
import software.amazon.awssdk.core.flow.ResponseIterator;
2123
import software.amazon.awssdk.core.regions.Region;
2224
import software.amazon.awssdk.http.nio.netty.NettySdkHttpClientFactory;
23-
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
24-
import software.amazon.awssdk.services.kinesis.KinesisAsyncClientBuilder;
25+
import software.amazon.awssdk.services.kinesis.model.DescribeStreamConsumerRequest;
2526
import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
2627
import software.amazon.awssdk.services.kinesis.model.RecordBatchEvent;
28+
import software.amazon.awssdk.services.kinesis.model.RegisterStreamConsumerResponse;
2729
import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
2830
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardRequest;
2931
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardResponse;
3032

3133
public class H2Demo {
3234

33-
private static final String ALPHA_STREAM_NAME = "foobar";
34-
private static final String DEVPERF_STREAM_NAME = "prashray-50";
35+
private static final String STREAM_NAME = "foobar";
36+
private static final String CONSUMER_ARN = "arn:aws:kinesis:us-east-1:052958737983:stream/foobar/consumer/consumer1:123456";
3537

3638
public static final int COUNT = 500_000;
3739
public static final int INTERVAL = 10;
3840
private static final Random random = new Random();
3941

4042
public static void main(String[] args) throws InterruptedException, UnsupportedEncodingException {
4143
BasicConfigurator.configure();
44+
// KinesisAsyncClient client = alpha(
45+
// KinesisAsyncClient
46+
// .builder()
47+
// .asyncHttpConfiguration(ClientAsyncHttpConfiguration
48+
// .builder()
49+
// .httpClientFactory(NettySdkHttpClientFactory.builder()
50+
// .trustAllCertificates(true)
51+
// .build())
52+
// .build())
53+
// ).build();
54+
55+
int numSubscribers = 1;
56+
CountDownLatch latch = new CountDownLatch(numSubscribers);
4257
KinesisAsyncClient client = alpha(
4358
KinesisAsyncClient
4459
.builder()
4560
.asyncHttpConfiguration(ClientAsyncHttpConfiguration
4661
.builder()
4762
.httpClientFactory(NettySdkHttpClientFactory.builder()
4863
.trustAllCertificates(true)
49-
.maxConnectionsPerEndpoint(10)
5064
.build())
5165
.build())
5266
).build();
67+
String streamArn = client.describeStream(r -> r.streamName(STREAM_NAME))
68+
.join().streamDescription().streamARN();
69+
client.describeStreamConsumer(DescribeStreamConsumerRequest.builder()
70+
.consumerName("java-sdk-consumer")
71+
.streamARN(streamArn)
72+
.build());
73+
if(true) {
74+
System.exit(0);
75+
}
76+
// String streamArn = client.describeStream(r -> r.streamName(STREAM_NAME)).join().streamDescription().streamARN();
77+
String consumerArn = client.describeStreamConsumer(r -> r.streamARN(streamArn)
78+
.consumerName("shorea-consumer"))
79+
.join().consumerDescription().consumerARN();
80+
81+
ExecutorService recordProducer = startProducer(client);
82+
ExecutorService subscriberExecutor = Executors.newFixedThreadPool(numSubscribers);
83+
for (int i = 1; i <= numSubscribers; i++) {
84+
int streamNum = i;
85+
subscriberExecutor.submit(() -> {
86+
try {
87+
subscribeToShardResponseHandler(client, "Stream-" + streamNum, consumerArn).join();
88+
89+
// ResponseIterator<SubscribeToShardResponse, RecordBatchEvent> iterator =
90+
// client.subscribeToShardBlocking(SubscribeToShardRequest.builder()
91+
// .consumerARN(STREAM_NAME)
92+
// .shardId("shardId-000000000000")
93+
// .shardIteratorType(ShardIteratorType.LATEST)
94+
// .consumerARN(consumerArn)
95+
// .build());
96+
// System.out.println("Has iterator");
97+
// iterator.forEachRemaining(System.out::println);
98+
System.out.println("Finished processing for stream " + streamNum);
99+
System.out.println("Closing client");
100+
} catch (Exception e) {
101+
e.printStackTrace();
102+
System.out.println("Stream " + streamNum + " failed");
103+
} finally {
104+
latch.countDown();
105+
}
106+
107+
});
108+
}
109+
latch.await();
110+
System.out.println("Closing client");
111+
client.close();
112+
System.out.println("All subscribers finished");
113+
subscriberExecutor.shutdown();
114+
subscriberExecutor.awaitTermination(1000, TimeUnit.SECONDS);
53115

54-
// ExecutorService recordProducer = startProducer();
55116
long start = System.nanoTime();
56-
ResponseIterator<SubscribeToShardResponse, RecordBatchEvent> iterator =
57-
client.subscribeToShardBlocking(SubscribeToShardRequest.builder()
58-
.consumerARN(ALPHA_STREAM_NAME)
59-
.shardId("shardId-000000000000")
60-
.shardIteratorType(ShardIteratorType.LATEST)
61-
.streamName(ALPHA_STREAM_NAME)
62-
.build());
63-
iterator.forEachRemaining(System.out::println);
64-
// subscribeToShardResponseHandler(client).join();
117+
// ResponseIterator<SubscribeToShardResponse, RecordBatchEvent> iterator =
118+
// client.subscribeToShardBlocking(SubscribeToShardRequest.builder()
119+
// .consumerARN(ALPHA_STREAM_NAME)
120+
// .shardId("shardId-000000000000")
121+
// .shardIteratorType(ShardIteratorType.LATEST)
122+
// .streamName(ALPHA_STREAM_NAME)
123+
// .build());
124+
// iterator.forEachRemaining(System.out::println);
65125
try {
66126
System.out.println("Total time = " + (System.nanoTime() - start));
67127
} catch (Exception e) {
68128
System.out.println("Closing client");
69-
client.close();
70129
}
71-
System.out.println("Closing client");
72-
client.close();
73-
// recordProducer.shutdownNow();
130+
recordProducer.shutdownNow();
74131
}
75132

76-
private static ExecutorService startProducer() {
77-
KinesisAsyncClient client = alpha(
78-
KinesisAsyncClient
79-
.builder()
80-
.asyncHttpConfiguration(ClientAsyncHttpConfiguration
81-
.builder()
82-
.httpClientFactory(NettySdkHttpClientFactory.builder()
83-
.trustAllCertificates(true)
84-
.maxConnectionsPerEndpoint(10)
85-
.build())
86-
.build())
87-
).build();
133+
private static ExecutorService startProducer(KinesisAsyncClient client) {
88134
ExecutorService recordProducer = Executors.newSingleThreadExecutor();
89135
recordProducer.submit(() -> {
90136
while (true) {
137+
System.out.println("Putting record");
91138
client.putRecord(PutRecordRequest.builder()
92-
.streamName(ALPHA_STREAM_NAME)
139+
.streamName(STREAM_NAME)
93140
.partitionKey(UUID.randomUUID().toString())
94-
.data(ByteBuffer.wrap(randomBytes(100)))
141+
.data(ByteBuffer.wrap(randomBytes(1000 * 100)))
95142
.build())
96143
.join();
97144
try {
@@ -107,19 +154,19 @@ private static ExecutorService startProducer() {
107154
return recordProducer;
108155
}
109156

110-
private static CompletableFuture<Integer> subscribeToShardResponseHandler(KinesisAsyncClient client) {
157+
private static CompletableFuture<Integer> subscribeToShardResponseHandler(KinesisAsyncClient client, String prefix, String consumerArn) {
111158
return client.subscribeToShard(SubscribeToShardRequest.builder()
112-
.consumerARN(ALPHA_STREAM_NAME)
159+
.consumerARN(CONSUMER_ARN)
113160
.shardId("shardId-000000000000")
114161
.shardIteratorType(ShardIteratorType.LATEST)
115-
.streamName(ALPHA_STREAM_NAME)
162+
.consumerARN(consumerArn)
116163
.build(),
117164
new FlowResponseTransformer<SubscribeToShardResponse, RecordBatchEvent, Integer>() {
118165
AtomicInteger count = new AtomicInteger(0);
119166

120167
@Override
121168
public void responseReceived(SubscribeToShardResponse response) {
122-
System.out.println("Initial Response = " + response);
169+
System.out.println(prefix + ": Initial Response = " + response);
123170
}
124171

125172
@Override
@@ -133,7 +180,7 @@ public void onSubscribe(Subscription subscription) {
133180
@Override
134181
public void onNext(RecordBatchEvent recordBatchEvent) {
135182
count.incrementAndGet();
136-
System.out.println("RECORDS = " + recordBatchEvent);
183+
// System.out.println(prefix + ": Records = " + recordBatchEvent);
137184
}
138185

139186
@Override
@@ -184,4 +231,10 @@ private static KinesisAsyncClientBuilder devPerf(KinesisAsyncClientBuilder build
184231
.credentialsProvider(ProfileCredentialsProvider.create("kinesis-dev-perf"));
185232
}
186233

234+
private static KinesisAsyncClientBuilder hailstone(KinesisAsyncClientBuilder builder) {
235+
return builder.endpointOverride(URI.create("https://kinesis-hailstoneperf.pdx.amazon.com"))
236+
.region(Region.US_EAST_1)
237+
.credentialsProvider(ProfileCredentialsProvider.create("kinesis-hailstone"));
238+
}
239+
187240
}

0 commit comments

Comments
 (0)