Add assume role support (#45)

Co-authored-by: German Osin <german.osin@gmail.com>

Author: shbodya

src/main/java/io/kafbat/ui/serde/glue/GlueSerde.java

@@ -55,6 +55,8 @@
 import software.amazon.awssdk.services.glue.model.GetSchemaVersionResponse;
 import software.amazon.awssdk.services.glue.model.SchemaId;
 import software.amazon.awssdk.services.glue.model.SchemaVersionNumber;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
 
 public class GlueSerde implements Serde {
 
@@ -88,6 +90,7 @@ public void configure(PropertyResolver serdeProperties,
         serdeProperties.getProperty("region", String.class)
             .orElseThrow(() -> new IllegalArgumentException("region not provided for GlueSerde")),
         serdeProperties.getProperty("endpoint", String.class).orElse(null),
+        serdeProperties.getProperty("roleArn", String.class).orElse(null),
         serdeProperties.getProperty("registry", String.class)
             .orElseThrow(() -> new IllegalArgumentException("registry not provided for GlueSerde")),
         serdeProperties.getProperty("keySchemaNameTemplate", String.class)
@@ -114,6 +117,7 @@ public void configure(PropertyResolver serdeProperties,
   void configure(AwsCredentialsProvider credentialsProvider,
                  String region,
                  @Nullable String endpoint,
+                 @Nullable String roleArn,
                  String registryName,
                  @Nullable String keySchemaNameTemplate,
                  String valueSchemaNameTemplate,
@@ -160,6 +164,19 @@ static AwsCredentialsProvider createCredentialsProvider(PropertyResolver serdePr
           .orElseGet(() -> () -> AwsBasicCredentials.create(awsAccessKey.get(), awsSecretKey.get()));
     }
 
+    Optional<String> roleArn = serdeProperties.getProperty("roleArn", String.class);
+    if (roleArn.isPresent()) {
+      return StsAssumeRoleCredentialsProvider.builder()
+          .refreshRequest(b -> b.roleArn(roleArn.get())
+              .roleSessionName("kafbat-ui-" + UUID.randomUUID()))
+          .stsClient(StsClient.builder()
+              .credentialsProvider(DefaultCredentialsProvider.create())
+              .region(Region.of(serdeProperties.getProperty("region", String.class)
+                  .orElseThrow(() -> new IllegalArgumentException("region required for assume role"))))
+              .build())
+          .build();
+    }
+
     Optional<String> profileName = serdeProperties.getProperty("awsProfileName", String.class);
     Optional<String> profileFile = serdeProperties.getProperty("awsProfileFile", String.class);
     if (profileName.isPresent() || profileFile.isPresent()) {

src/test/java/io/kafbat/ui/serde/glue/GlueSerdeTest.java

@@ -83,8 +83,9 @@ class GlueSerdeTest {
 
   private static final String REGISTRY_NAME = "kui-glue-serde-test-registry";
 
+  private static final Network NETWORK = Network.newNetwork();
   private static final KafkaContainer KAFKA = new KafkaContainer(
-      DockerImageName.parse("confluentinc/cp-kafka:7.2.1")).withNetwork(Network.SHARED);
+      DockerImageName.parse("confluentinc/cp-kafka:7.2.1")).withNetwork(NETWORK);
 
   private static GlueClient GLUE_CLIENT;
 
@@ -115,6 +116,7 @@ static void checkCredsResolving() {
   @AfterAll
   static void tearDown() {
     KAFKA.close();
+    NETWORK.close();
     try {
       GLUE_CLIENT.deleteRegistry(
           DeleteRegistryRequest.builder().registryId(
@@ -194,6 +196,7 @@ private <T> void checkDeserializerIsCompatibleWithKafkaLibrarySerializer(
           DefaultCredentialsProvider.create(),
           REGION,
           null,
+          null,
           REGISTRY_NAME,
           null,
           "%s",
@@ -228,6 +231,7 @@ private <T> void checkSerializerIsCompatibleWithKafkaLibraryDeserializer(
           DefaultCredentialsProvider.create(),
           REGION,
           null,
+          null,
           REGISTRY_NAME,
           null,
           "%s",
@@ -403,6 +407,7 @@ void canSerializeAndCanDeserializeCheckTopicSchemaMappingMap() {
           DefaultCredentialsProvider.create(),
           REGION,
           null,
+          null,
           REGISTRY_NAME,
           null,
           "%s",
@@ -457,6 +462,7 @@ void canSerializeAndCanDeserializeUsesTopicKVTemplateToFindSchemas() {
           DefaultCredentialsProvider.create(),
           REGION,
           null,
+          null,
           REGISTRY_NAME,
           "%s-key",
           "%s-value",
@@ -484,6 +490,7 @@ void canDeserializeReturnsTrueForAnyTopicIfSchemaExistenceCheckIsDisabled() {
           DefaultCredentialsProvider.create(),
           REGION,
           null,
+          null,
           REGISTRY_NAME,
           "%s-key",
           "%s-value",