This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 7a4a46b824319556bd55f670b546cd1dd22efad1
Author: Andrea Cosentino <anco...@gmail.com>
AuthorDate: Fri Oct 21 13:53:03 2022 +0200
CAMEL-18131 - camel-health - Add health checks for components that has
extension for connectivity verification - AWS Kinesis
Signed-off-by: Andrea Cosentino <anco...@gmail.com>
---
...inesis2ConsumerHealthCheckProfileCredsTest.java | 106 ++++++++++++++++++++
...Kinesis2ConsumerHealthCheckStaticCredsTest.java | 107 +++++++++++++++++++++
.../Kinesis2ConsumerHealthCustomClientTest.java | 106 ++++++++++++++++++++
3 files changed, 319 insertions(+)
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ConsumerHealthCheckProfileCredsTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ConsumerHealthCheckProfileCredsTest.java
new file mode 100644
index 00000000000..027815735d9
--- /dev/null
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ConsumerHealthCheckProfileCredsTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckHelper;
+import org.apache.camel.health.HealthCheckRegistry;
+import org.apache.camel.impl.health.DefaultHealthCheckRegistry;
+import org.apache.camel.test.infra.aws.common.services.AWSService;
+import org.apache.camel.test.infra.aws2.clients.AWSSDKClientUtils;
+import org.apache.camel.test.infra.aws2.services.AWSServiceFactory;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.concurrent.TimeUnit;
+
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
+
+public class Kinesis2ConsumerHealthCheckProfileCredsTest extends
CamelTestSupport {
+
+ @RegisterExtension
+ public static AWSService service = AWSServiceFactory.createS3Service();
+
+ private static final Logger LOG =
LoggerFactory.getLogger(Kinesis2ConsumerHealthCheckProfileCredsTest.class);
+
+ CamelContext context;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ context = super.createCamelContext();
+ context.getPropertiesComponent().setLocation("ref:prop");
+ Kinesis2Component component = new Kinesis2Component(context);
+
component.getConfiguration().setAmazonKinesisClient(AWSSDKClientUtils.newKinesisClient());
+ component.init();
+ context.addComponent("aws2-kinesis", component);
+
+ HealthCheckRegistry registry = new DefaultHealthCheckRegistry();
+ registry.setCamelContext(context);
+ Object hc = registry.resolveById("context");
+ registry.register(hc);
+ hc = registry.resolveById("routes");
+ registry.register(hc);
+ hc = registry.resolveById("consumers");
+ registry.register(hc);
+ context.setExtension(HealthCheckRegistry.class, registry);
+
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() {
+
from("aws2-kinesis://stream?region=l&useDefaultCredentialsProvider=true")
+
.startupOrder(2).log("${body}").routeId("test-health-it");
+ }
+ };
+ }
+
+ @Test
+ public void testConnectivity() {
+
+ Collection<HealthCheck.Result> res =
HealthCheckHelper.invokeLiveness(context);
+ boolean up = res.stream().allMatch(r ->
r.getState().equals(HealthCheck.State.UP));
+ Assertions.assertTrue(up, "liveness check");
+
+ // health-check readiness should be down
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+ Collection<HealthCheck.Result> res2 =
HealthCheckHelper.invokeReadiness(context);
+ boolean down = res2.stream().allMatch(r ->
r.getState().equals(HealthCheck.State.DOWN));
+ boolean containsKinesis2HealthCheck = res2.stream()
+ .filter(result ->
result.getCheck().getId().startsWith("aws2-kinesis-consumer"))
+ .findAny()
+ .isPresent();
+ boolean hasRegionMessage = res2.stream()
+ .anyMatch(r -> r.getMessage().stream().anyMatch(msg ->
msg.contains("region")));
+ Assertions.assertTrue(down, "liveness check");
+ Assertions.assertTrue(containsKinesis2HealthCheck, "aws2-kinesis
check");
+ Assertions.assertTrue(hasRegionMessage, "aws2-kinesis check error
message");
+ });
+
+ }
+}
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ConsumerHealthCheckStaticCredsTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ConsumerHealthCheckStaticCredsTest.java
new file mode 100644
index 00000000000..c7a8df6823f
--- /dev/null
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ConsumerHealthCheckStaticCredsTest.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckHelper;
+import org.apache.camel.health.HealthCheckRegistry;
+import org.apache.camel.impl.health.DefaultHealthCheckRegistry;
+import org.apache.camel.test.infra.aws.common.services.AWSService;
+import org.apache.camel.test.infra.aws2.clients.AWSSDKClientUtils;
+import org.apache.camel.test.infra.aws2.services.AWSServiceFactory;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.concurrent.TimeUnit;
+
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
+
+public class Kinesis2ConsumerHealthCheckStaticCredsTest extends
CamelTestSupport {
+
+ @RegisterExtension
+ public static AWSService service = AWSServiceFactory.createS3Service();
+
+ private static final Logger LOG =
LoggerFactory.getLogger(Kinesis2ConsumerHealthCheckStaticCredsTest.class);
+
+ CamelContext context;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ context = super.createCamelContext();
+ context.getPropertiesComponent().setLocation("ref:prop");
+ Kinesis2Component component = new Kinesis2Component(context);
+
component.getConfiguration().setAmazonKinesisClient(AWSSDKClientUtils.newKinesisClient());
+ component.init();
+ context.addComponent("aws2-kinesis", component);
+
+ // install health check manually (yes a bit cumbersome)
+ HealthCheckRegistry registry = new DefaultHealthCheckRegistry();
+ registry.setCamelContext(context);
+ Object hc = registry.resolveById("context");
+ registry.register(hc);
+ hc = registry.resolveById("routes");
+ registry.register(hc);
+ hc = registry.resolveById("consumers");
+ registry.register(hc);
+ context.setExtension(HealthCheckRegistry.class, registry);
+
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() {
+ from("aws2-kinesis://stream?region=l&secretKey=l&accessKey=k")
+
.startupOrder(2).log("${body}").routeId("test-health-it");
+ }
+ };
+ }
+
+ @Test
+ public void testConnectivity() {
+
+ Collection<HealthCheck.Result> res =
HealthCheckHelper.invokeLiveness(context);
+ boolean up = res.stream().allMatch(r ->
r.getState().equals(HealthCheck.State.UP));
+ Assertions.assertTrue(up, "liveness check");
+
+ // health-check readiness should be down
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+ Collection<HealthCheck.Result> res2 =
HealthCheckHelper.invokeReadiness(context);
+ boolean down = res2.stream().allMatch(r ->
r.getState().equals(HealthCheck.State.DOWN));
+ boolean containsKinesis2HealthCheck = res2.stream()
+ .filter(result ->
result.getCheck().getId().startsWith("aws2-kinesis-consumer"))
+ .findAny()
+ .isPresent();
+ boolean hasRegionMessage = res2.stream()
+ .anyMatch(r -> r.getMessage().stream().anyMatch(msg ->
msg.contains("region")));
+ Assertions.assertTrue(down, "liveness check");
+ Assertions.assertTrue(containsKinesis2HealthCheck, "aws2-kinesis
check");
+ Assertions.assertTrue(hasRegionMessage, "aws2-kinesis check error
message");
+ });
+
+ }
+}
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ConsumerHealthCustomClientTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ConsumerHealthCustomClientTest.java
new file mode 100644
index 00000000000..0d0bd9fd6ab
--- /dev/null
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ConsumerHealthCustomClientTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckHelper;
+import org.apache.camel.health.HealthCheckRegistry;
+import org.apache.camel.impl.health.DefaultHealthCheckRegistry;
+import org.apache.camel.test.infra.aws.common.services.AWSService;
+import org.apache.camel.test.infra.aws2.clients.AWSSDKClientUtils;
+import org.apache.camel.test.infra.aws2.services.AWSServiceFactory;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.concurrent.TimeUnit;
+
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
+
+public class Kinesis2ConsumerHealthCustomClientTest extends CamelTestSupport {
+
+ @RegisterExtension
+ public static AWSService service = AWSServiceFactory.createS3Service();
+
+ private static final Logger LOG =
LoggerFactory.getLogger(Kinesis2ConsumerHealthCustomClientTest.class);
+
+ CamelContext context;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ context = super.createCamelContext();
+ context.getPropertiesComponent().setLocation("ref:prop");
+ Kinesis2Component component = new Kinesis2Component(context);
+
component.getConfiguration().setAmazonKinesisClient(AWSSDKClientUtils.newKinesisClient());
+ component.init();
+ context.addComponent("aws2-kinesis", component);
+
+ HealthCheckRegistry registry = new DefaultHealthCheckRegistry();
+ registry.setCamelContext(context);
+ Object hc = registry.resolveById("context");
+ registry.register(hc);
+ hc = registry.resolveById("routes");
+ registry.register(hc);
+ hc = registry.resolveById("consumers");
+ registry.register(hc);
+ context.setExtension(HealthCheckRegistry.class, registry);
+
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() {
+ from("aws2-kinesis://stream")
+
.startupOrder(2).log("${body}").routeId("test-health-it");
+ }
+ };
+ }
+
+ @Test
+ public void testConnectivity() {
+
+ Collection<HealthCheck.Result> res =
HealthCheckHelper.invokeLiveness(context);
+ boolean up = res.stream().allMatch(r ->
r.getState().equals(HealthCheck.State.UP));
+ Assertions.assertTrue(up, "liveness check");
+
+ // health-check readiness should be down
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+ Collection<HealthCheck.Result> res2 =
HealthCheckHelper.invokeReadiness(context);
+ boolean down = res2.stream().allMatch(r ->
r.getState().equals(HealthCheck.State.DOWN));
+ boolean containsAws2S3HealthCheck = res2.stream()
+ .filter(result ->
result.getCheck().getId().startsWith("aws2-kinesis-consumer"))
+ .findAny()
+ .isPresent();
+ boolean hasRegionMessage = res2.stream()
+ .anyMatch(r -> r.getMessage().stream().anyMatch(msg ->
msg.contains("region")));
+ Assertions.assertTrue(down, "liveness check");
+ Assertions.assertTrue(containsAws2S3HealthCheck, "aws2-kinesis
check");
+ Assertions.assertFalse(hasRegionMessage, "aws2-kinesis check error
message");
+ });
+
+ }
+}
