iSharkFly-Docs
/
java-tutorials
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

JAVA-15980 Upgrade AWS Kinesis (#13294) 

* JAVA-15980 Upgrade AWS Kinesis

* JAVA-15980 Minor readability improvement
This commit is contained in:
Anastasios Ioannidis 2023-02-03 19:47:48 +02:00 committed by GitHub
parent bf69ecf4bd
commit 456ff7d029
8 changed files with 71 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/pom.xml
View File

@ -36,7 +36,7 @@
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>amazon-kinesis-client</artifactId>
<version>1.11.2</version>
<version>1.14.9</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
@ -46,9 +46,9 @@
</dependencies>
<properties>
<aws-sdk.version>1.11.632</aws-sdk.version>
<spring-cloud-stream-kinesis-binder.version>2.0.2.RELEASE</spring-cloud-stream-kinesis-binder.version>
<spring-cloud-stream-test.version>2.2.1.RELEASE</spring-cloud-stream-test.version>
<aws-sdk.version>1.12.380</aws-sdk.version>
<spring-cloud-stream-kinesis-binder.version>2.2.0</spring-cloud-stream-kinesis-binder.version>
<spring-cloud-stream-test.version>4.0.0</spring-cloud-stream-test.version>
</properties>
</project>

22
spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/binder/ConsumerBinder.java
View File

@ -1,16 +1,16 @@
package com.baeldung.binder;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;
import java.util.function.Consumer;
@Component
@EnableBinding(Sink.class)
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ConsumerBinder {
@StreamListener(Sink.INPUT)
public void consume(String ip) {
System.out.println(ip);
}
@Bean
Consumer<String> input() {
return str -> {
System.out.println(str);
};
}
}

6
spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/binder/KinesisBinderApplication.java
View File

@ -6,7 +6,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class KinesisBinderApplication {
public static void main(String[] args) {
SpringApplication.run(KinesisBinderApplication.class, args);
}
public static void main(String[] args) {
SpringApplication.run(KinesisBinderApplication.class, args);
}
}

26
spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/binder/ProducerBinder.java
View File

@ -1,24 +1,20 @@
package com.baeldung.binder;
import java.util.function.Supplier;
import java.util.stream.IntStream;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@Component
@EnableBinding(Source.class)
public class ProducerBinder {
@Configuration
class ProducerBinder {
@Autowired
private Source source;
@Scheduled(fixedDelay = 3000L)
private void produce() {
IntStream.range(1, 200).mapToObj(ipSuffix -> "192.168.0." + ipSuffix)
.forEach(entry -> source.output().send(MessageBuilder.withPayload(entry).build()));
@Bean
public Supplier output() {
return () -> IntStream.range(1, 200)
.mapToObj(ipSuffix -> "192.168.0." + ipSuffix)
.map(entry -> MessageBuilder.withPayload(entry)
.build());
}
}

39
spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/kclkpl/KinesisKCLApplication.java
View File

@ -1,11 +1,14 @@
package com.baeldung.kclkpl;
import static com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.regions.Regions;
@ -17,32 +20,30 @@ public class KinesisKCLApplication implements ApplicationRunner {
@Value("${aws.access.key}")
private String accessKey;
@Value("${aws.secret.key}")
private String secretKey;
@Value("${ips.stream}")
private String IPS_STREAM;
public static void main(String[] args) {
SpringApplication.run(KinesisKCLApplication.class, args);
}
@Override
public void run(ApplicationArguments args) throws Exception {
BasicAWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secretKey);
KinesisClientLibConfiguration consumerConfig = new KinesisClientLibConfiguration(
"KinesisKCLConsumer",
IPS_STREAM,
new AWSStaticCredentialsProvider(awsCredentials),
"KinesisKCLConsumer")
.withRegionName(Regions.EU_CENTRAL_1.getName());
@Override
public void run(ApplicationArguments args) throws Exception {
BasicAWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secretKey);
new Worker.Builder()
.recordProcessorFactory(new IpProcessorFactory())
.config(consumerConfig)
.build()
.run();
}
KinesisClientLibConfiguration consumerConfig = new KinesisClientLibConfiguration("KinesisKCLConsumer", IPS_STREAM, "", "", DEFAULT_INITIAL_POSITION_IN_STREAM, new AWSStaticCredentialsProvider(awsCredentials),
new AWSStaticCredentialsProvider(awsCredentials), new AWSStaticCredentialsProvider(awsCredentials), DEFAULT_FAILOVER_TIME_MILLIS, "KinesisKCLConsumer", DEFAULT_MAX_RECORDS, DEFAULT_IDLETIME_BETWEEN_READS_MILLIS,
DEFAULT_DONT_CALL_PROCESS_RECORDS_FOR_EMPTY_RECORD_LIST, DEFAULT_PARENT_SHARD_POLL_INTERVAL_MILLIS, DEFAULT_SHARD_SYNC_INTERVAL_MILLIS, DEFAULT_CLEANUP_LEASES_UPON_SHARDS_COMPLETION, new ClientConfiguration(), new ClientConfiguration(),
new ClientConfiguration(), DEFAULT_TASK_BACKOFF_TIME_MILLIS, DEFAULT_METRICS_BUFFER_TIME_MILLIS, DEFAULT_METRICS_MAX_QUEUE_SIZE, DEFAULT_VALIDATE_SEQUENCE_NUMBER_BEFORE_CHECKPOINTING, Regions.EU_CENTRAL_1.getName(), DEFAULT_SHUTDOWN_GRACE_MILLIS,
DEFAULT_DDB_BILLING_MODE, null, 0, 0, 0);
new Worker.Builder().recordProcessorFactory(new IpProcessorFactory())
.config(consumerConfig)
.build()
.run();
}
}

17
spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/kclkpl/KinesisKPLApplication.java
View File

@ -16,23 +16,22 @@ public class KinesisKPLApplication {
@Value("${aws.access.key}")
private String accessKey;
@Value("${aws.secret.key}")
private String secretKey;
public static void main(String[] args) {
SpringApplication.run(KinesisKPLApplication.class, args);
}
@Bean
public KinesisProducer kinesisProducer() {
BasicAWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secretKey);
KinesisProducerConfiguration producerConfig = new KinesisProducerConfiguration()
.setCredentialsProvider(new AWSStaticCredentialsProvider(awsCredentials))
.setVerifyCertificate(false)
.setRegion(Regions.EU_CENTRAL_1.getName());
BasicAWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secretKey);
KinesisProducerConfiguration producerConfig = new KinesisProducerConfiguration().setCredentialsProvider(new AWSStaticCredentialsProvider(awsCredentials))
.setVerifyCertificate(false)
.setRegion(Regions.EU_CENTRAL_1.getName());
return new KinesisProducer(producerConfig);
return new KinesisProducer(producerConfig);
}
}

12
spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/resources/application.properties
View File

@ -12,9 +12,11 @@ cloud.aws.credentials.secret-key=my-aws-secret-key
cloud.aws.region.static=eu-central-1
cloud.aws.stack.auto=false
spring.cloud.stream.bindings.input.destination=live-ips
spring.cloud.stream.bindings.input.group=live-ips-group
spring.cloud.stream.bindings.input.content-type=text/plain
spring.cloud.stream.bindings.input-in-0.destination=live-ips
spring.cloud.stream.bindings.input-in-0.group=live-ips-group
spring.cloud.stream.bindings.input-in-0.content-type=text/plain
spring.cloud.stream.function.definition = input
spring.cloud.stream.bindings.output.destination=myStream
spring.cloud.stream.bindings.output.content-type=text/plain
spring.cloud.stream.bindings.output-out-0.destination=myStream
spring.cloud.stream.bindings.output-out-0.content-type=text/plain
spring.cloud.stream.poller.fixed-delay = 3000

12
spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/resources/application.properties
View File

@ -12,9 +12,11 @@ cloud.aws.credentials.secret-key=my-aws-secret-key
cloud.aws.region.static=eu-central-1
cloud.aws.stack.auto=false
spring.cloud.stream.bindings.input.destination=live-ips
spring.cloud.stream.bindings.input.group=live-ips-group
spring.cloud.stream.bindings.input.content-type=text/plain
spring.cloud.stream.bindings.input-in-0.destination=live-ips
spring.cloud.stream.bindings.input-in-0.group=live-ips-group
spring.cloud.stream.bindings.input-in-0.content-type=text/plain
spring.cloud.stream.function.definition = input
spring.cloud.stream.bindings.output.destination=myStream
spring.cloud.stream.bindings.output.content-type=text/plain
spring.cloud.stream.bindings.output-out-0.destination=myStream
spring.cloud.stream.bindings.output-out-0.content-type=text/plain
spring.cloud.stream.poller.fixed-delay = 3000