Validate streaming information over Amazon MSK utilizing schemas in cross-account AWS Glue Schema Registry

Immediately’s companies face an unprecedented progress within the quantity of information. A rising portion of the info is generated in actual time by IoT gadgets, web sites, enterprise purposes, and numerous different sources. Companies must course of and analyze this information as quickly because it arrives to make enterprise choices in actual time. Amazon Managed Streaming for Apache Kafka (Amazon MSK) is a totally managed service that allows constructing and operating stream processing purposes that use Apache Kafka to gather and course of information in actual time.

Stream processing purposes utilizing Apache Kafka don’t talk with one another immediately; they convey through sending and receiving messages over Kafka subjects. For stream processing purposes to speak effectively and confidently, a message payload construction should be outlined by way of attributes and information varieties. This construction describes the schema purposes use when sending and receiving messages. Nonetheless, with a lot of producer and shopper purposes, even a small change in schema (eradicating a discipline, including a brand new discipline, or change in information kind) could trigger points for downstream purposes which can be troublesome to debug and repair.

Historically, groups have relied on change administration processes (equivalent to approvals and upkeep home windows) or different casual mechanisms (documentation, emails, collaboration instruments, and so forth) to tell each other of information schema modifications. Nonetheless, these mechanisms don’t scale and are susceptible to errors. The AWS Glue Schema Registry means that you can centrally publish, uncover, management, validate, and evolve schemas for stream processing purposes. With the AWS Glue Schema Registry, you’ll be able to handle and implement schemas on information streaming purposes utilizing Apache Kafka, Amazon MSK, Amazon Kinesis Knowledge Streams, Amazon Kinesis Knowledge Analytics for Apache Flink, and AWS Lambda.

This publish demonstrates how Apache Kafka stream processing purposes validate messages utilizing an Apache Avro schema saved within the AWS Glue Schema registry residing in a central AWS account. We use the AWS Glue Schema Registry SerDe library and Avro SpecificRecord to validate messages in stream processing purposes whereas sending and receiving messages from a Kafka matter on an Amazon MSK cluster. Though we use an Avro schema for this publish, the identical method and idea applies to JSON schemas as nicely.

Use case

Let’s assume a fictitious rideshare firm that gives unicorn rides. To attract actionable insights, they should course of a stream of unicorn experience request messages. They count on rides to be highly regarded and need to be certain that their answer can scale. They’re additionally constructing a central information lake the place all their streaming and operation information is saved for evaluation. They’re buyer obsessed, in order that they count on so as to add new enjoyable options to future rides, like selecting the hair coloration of your unicorn, and might want to mirror these attributes within the experience request messages. To keep away from points in downstream purposes as a result of future schema modifications, they want a mechanism to validate messages with a schema hosted in a central schema registry. Having schemas in a central schema registry makes it simpler for the applying groups to publish, validate, evolve, and keep schemas in a single place.

Resolution overview

The corporate makes use of Amazon MSK to seize and distribute the unicorn experience request messages at scale. They outline an Avro schema for unicorn experience requests as a result of it gives wealthy information buildings, helps direct mapping to JSON, in addition to a compact, quick, and binary information format. As a result of the schema was agreed prematurely, they determined to make use of Avro SpecificRecord.SpecificRecord is an interface from the Avro library that enables the usage of an Avro document as a POJO. That is completed by producing a Java class (or lessons) from the schema, through the use of avro-maven-plugin. They use AWS Identification and Entry Administration (IAM) cross-account roles to permit producer and shopper purposes from the opposite AWS account to soundly and securely entry schemas within the central Schema Registry account.

The AWS Glue Schema Registry is in Account B, whereas the MSK cluster and Kafka producer and shopper purposes are in Account A. We use the next two IAM roles to allow cross-account entry to the AWS Glue Schema Registry. Apache Kafka shoppers in Account A assume a job in Account B utilizing an identity-based coverage as a result of the AWS Glue Schema Registry doesn’t help resource-based insurance policies.

• Account A IAM position – Permits producer and shopper purposes to imagine an IAM position in Account B.
• Account B IAM position – Trusts all IAM principals from Account A and permits them to carry out learn actions on the AWS Glue Schema Registry in Account B. In an actual use case state of affairs, IAM principals that may assume cross-account roles needs to be scoped extra particularly.

The next structure diagram illustrates the answer:

The answer works as follows:

1. A Kafka producer operating in Account A assumes the cross-account Schema Registry IAM position in Account B by calling the AWS Safety Token Service (AWS STS) assumeRole API.
2. The Kafka producer retrieves the unicorn experience request Avro schema model ID from the AWS Glue Schema Registry for the schema that’s embedded within the unicorn experience request POJO. Fetching the schema model ID is internally managed by the AWS Glue Schema Registry SerDe’s serializer. The serializer must be configured as a part of the Kafka producer configuration.
3. If the schema exists within the AWS Glue Schema Registry, the serializer decorates the info document with the schema model ID after which serializes it earlier than delivering it to the Kafka matter on the MSK cluster.
4. The Kafka shopper operating in Account A assumes the cross-account Schema Registry IAM position in Account B by calling the AWS STS assumeRole API.
5. The Kafka shopper begins polling the Kafka matter on the MSK cluster for information information.
6. The Kafka shopper retrieves the unicorn experience request Avro schema from the AWS Glue Schema Registry, matching the schema model ID that’s encoded within the unicorn experience request information document. Fetching the schema is internally managed by the AWS Glue Schema Registry SerDe’s deserializer. The deserializer must be configured as a part of the Kafka shopper configuration. If the schema exists within the AWS Glue Schema Registry, the deserializer deserializes the info document into the unicorn experience request POJO for the buyer to course of it.

The AWS Glue Schema Registry SerDe library additionally helps non-obligatory compression configuration to avoid wasting on information transfers. For extra details about the Schema Registry, see How the Schema Registry works.

Unicorn experience request Avro schema

The next schema (UnicornRideRequest.avsc) defines a document representing a unicorn experience request, which comprises experience request attributes together with the shopper attributes and system-recommended unicorn attributes:

{
    "kind": "document",
    "identify": "UnicornRideRequest",
    "namespace": "demo.glue.schema.registry.avro",
    "fields": [
      {"name": "request_id", "type": "int", "doc": "customer request id"},
      {"name": "pickup_address","type": "string","doc": "customer pickup address"},
      {"name": "destination_address","type": "string","doc": "customer destination address"},
      {"name": "ride_fare","type": "float","doc": "ride fare amount (USD)"},
      {"name": "ride_duration","type": "int","doc": "ride duration in minutes"},
      {"name": "preferred_unicorn_color","type": {"type": "enum","name": "UnicornPreferredColor","symbols": ["WHITE","BLACK","RED","BLUE","GREY"]}, "default": "WHITE"},
      {
        "identify": "recommended_unicorn",
        "kind": {
          "kind": "document",
          "identify": "RecommendedUnicorn",
          "fields": [
            {"name": "unicorn_id","type": "int", "doc": "recommended unicorn id"},
            {"name": "color","type": {"type": "enum","name": "unicorn_color","symbols": ["WHITE","RED","BLUE"]}},
            {"identify": "stars_rating", "kind": ["null", "int"], "default": null, "doc": "unicorn star scores primarily based on clients suggestions"}
          ]
        }
      },
      {
        "identify": "buyer",
        "kind": {
          "kind": "document",
          "identify": "Buyer",
          "fields": [
            {"name": "customer_account_no","type": "int", "doc": "customer account number"},
            {"name": "first_name","type": "string"},
            {"name": "middle_name","type": ["null","string"], "default": null},
            {"identify": "last_name","kind": "string"},
            {"identify": "email_addresses","kind": ["null", {"type":"array", "items":"string"}]},
            {"identify": "customer_address","kind": "string","doc": "buyer handle"},
            {"identify": "mode_of_payment","kind": {"kind": "enum","identify": "ModeOfPayment","symbols": ["CARD","CASH"]}, "default": "CARD"},
            {"identify": "customer_rating", "kind": ["null", "int"], "default": null}
          ]
        }
      }
    ]
  }

Stipulations

To make use of this answer, you need to have two AWS accounts:

• Account A – For the MSK cluster, Kafka producer and shopper Amazon Elastic Compute Cloud (Amazon EC2) situations, and AWS Cloud9 atmosphere
• Account B – For the Schema Registry and schema

For this answer, we use Area us-east-1, however you’ll be able to change this as per your necessities.

Subsequent, we create the sources in every account utilizing AWS CloudFormation templates.

Create sources in Account B

We create the next sources in Account B:

• A schema registry
• An Avro schema
• An IAM position with the AWSGlueSchemaRegistryReadonlyAccess managed coverage and an occasion profile, which permits all Account A IAM principals to imagine it
• The UnicornRideRequest.avsc Avro schema proven earlier, which is used as a schema definition within the CloudFormation template

Ensure you have the suitable permissions to create these sources.

1. Log in to Account B.
2. Launch the next CloudFormation stack.
3. For Stack identify, enter SchemaRegistryStack.
4. For Schema Registry identify, enter unicorn-ride-request-registry.
5. For Avro Schema identify, enter unicorn-ride-request-schema-avro.
6. For the Kafka shopper’s AWS account ID, enter your Account A ID.
7. For ExternalId, enter a singular random ID (for instance, demo10A), which needs to be offered by the Kafka shoppers in Account Some time assuming the IAM position on this account.

For extra details about cross-account safety, see The confused deputy drawback.

8. When the stack is full, on the Outputs tab of the stack, copy the worth for CrossAccountGlueSchemaRegistryRoleArn.

The Kafka producer and shopper purposes created in Account A assume this position to entry the Schema Registry and schema in Account B.

9. To confirm the sources had been created, on the AWS Glue console, select Schema registries within the navigation bar, and find unicorn-ride-request-registry.
   
10. Select the registry unicorn-ride-request-registry and confirm that it comprises unicorn-ride-request-schema-avro within the Schemas part.
11. Select the schema to see its content material.
    

The IAM position created by the SchemaRegistryStack stack permits all Account A IAM principals to imagine it and carry out learn actions on the AWS Glue Schema Registry. Let’s have a look at the belief relationships of the IAM position.

12. On the SchemaRegistryStack stack Outputs tab, copy the worth for CrossAccountGlueSchemaRegistryRoleName.
13. On the IAM console, seek for this position.
14. Select Belief relationships and have a look at its trusted entities to verify that Account A is listed.
15. Within the Circumstances part, verify that sts:ExternalId has the identical distinctive random ID offered throughout stack creation.

Create sources in Account A

We create the next sources in Account A:

• A VPC
• EC2 situations for the Kafka producer and shopper
• An AWS Cloud9 atmosphere
• An MSK cluster

As a prerequisite, create an EC2 keypair and obtain it in your machine to have the ability to SSH into EC2 situations. Additionally create an MSK cluster configuration with default values. It is advisable have permissions to create the CloudFormation stack, EC2 situations, AWS Cloud9 atmosphere, MSK cluster, MSK cluster configuration, and IAM position.

1. Log in to Account A.
2. Launch the next CloudFormation stack to launch the VPC, EC2 situations, and AWS Cloud9 atmosphere.
3. For Stack identify, enter MSKClientStack.
4. Present the VPC and subnet CIDR ranges.
5. For EC2 Keypair, select an present EC2 keypair.
6. For the newest EC2 AMI ID, choose the default choice.
7. For the cross-account IAM position ARN, use the worth for CrossAccountGlueSchemaRegistryRoleArn (out there on the Outputs tab of SchemaRegistryStack).
8. Await the stack to create efficiently.
9. Launch the next CloudFormation stack to create the MSK cluster.
10. For Stack identify, enter MSKClusterStack.
11. Use Amazon MSK model 2.7.1.
12. For the MSK cluster configuration ARN, enter the MSK cluster configuration ARN. One that you simply created as a part of the prerequisite.
13. For the MSK cluster configuration revision quantity, enter 1 or change it in keeping with your model.
14. For the shopper CloudFormation stack identify, enter MSKClientStack (the stack identify that you simply created previous to this stack).

Configure the Kafka producer

To configure the Kafka producer accessing the Schema Registry within the central AWS account, full the next steps:

1. Log in to Account A.
2. On the AWS Cloud9 console, select the Cloud9EC2Bastion atmosphere created by the MSKClientStack stack.
3. On the File menu, select Add Native Recordsdata.
4. Add the EC2 keypair file that you simply used earlier whereas creating the stack.
   
5. Open a brand new terminal and alter the EC2 keypair permissions:

   chmod 0400 <keypair PEM file>

6. SSH into the KafkaProducerInstance EC2 occasion and set the Area as per your requirement:

   ssh -i <keypair PEM file> ec2-user@<KafkaProducerInstance Non-public IP handle>
   aws configure set area <area>

7. Set the atmosphere variable MSK_CLUSTER_ARN pointing to the MSK cluster’s ARN:

   export MSK_CLUSTER_ARN=$(aws kafka list-clusters |  jq '.ClusterInfoList[] | choose (.ClusterName == "MSKClusterStack") | {ClusterArn} | be part of (" ")' | tr -d ")

Change the .ClusterName worth within the code in case you used a unique identify for the MSK cluster CloudFormation stack. The cluster identify is identical because the stack identify.

8. Set the atmosphere variable BOOTSTRAP_BROKERS pointing to the bootstrap brokers:

   export BOOTSTRAP_BROKERS=$(aws kafka get-bootstrap-brokers --cluster-arn $MSK_CLUSTER_ARN | jq -r .BootstrapBrokerString)

9. Confirm the atmosphere variables:

   echo $MSK_CLUSTER_ARN
   echo $BOOTSTRAP_BROKERS

10. Create a Kafka matter referred to as unicorn-ride-request-topic in your MSK cluster, which is utilized by the Kafka producer and shopper purposes later:

    cd ~/kafka
    
    ./bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_BROKERS 
    --topic unicorn-ride-request-topic 
    --create --partitions 3 --replication-factor 2
    
    ./bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_BROKERS --list

The MSKClientStack stack copied the Kafka producer shopper JAR file referred to as kafka-cross-account-gsr-producer.jar to the KafkaProducerInstance occasion. It comprises the Kafka producer shopper that sends messages to the Kafka matter unicorn-ride-request-topic on the MSK cluster and accesses the unicorn-ride-request-schema-avro Avro schema from the unicorn-ride-request-registry schema registry in Account B. The Kafka producer code, which we cowl later on this publish, is offered on GitHub.

1. Run the next instructions and confirm kafka-cross-account-gsr-producer.jar exists:
2. Run the next command to run the Kafka producer within the KafkaProducerInstance terminal:

   java -jar kafka-cross-account-gsr-producer.jar -bs $BOOTSTRAP_BROKERS 
   -rn <Account B IAM position arn that Kafka producer software must assume> 
   -topic unicorn-ride-request-topic 
   -reg us-east-1 
   -nm 500 
   -externalid <Account B IAM position exterior Id that you simply used whereas making a CF stack in Account B>

The code has the next parameters:

• -bs – $BOOTSTRAP_BROKERS (the MSK cluster bootstrap brokers)
• -rn – The CrossAccountGlueSchemaRegistryRoleArn worth from the SchemaRegistryStack stack outputs in Account B
• -topic – the Kafka matter unicorn-ride-request-topic
• -reg – us-east-1 (change it in keeping with your Area, it’s used for the AWS STS endpoint and Schema Registry)
• -nm: 500 (the variety of messages the producer software sends to the Kafka matter)
• -externalId – The identical exterior ID (for instance, demo10A) that you simply used whereas creating the CloudFormation stack in Account B

The next screenshot reveals the Kafka producer logs exhibiting Schema Model Id obtained..., which suggests it has retrieved the Avro schema unicorn-ride-request-schema-avro from Account B and messages had been despatched to the Kafka matter on the MSK cluster in Account A.

Kafka producer code

The whole Kafka producer implementation is offered on GitHub. On this part, we break down the code.

• getProducerConfig() initializes the producer properties, as proven within the following code:
  • VALUE_SERIALIZER_CLASS_CONFIG – The GlueSchemaRegistryKafkaSerializer.class.getName() AWS serializer implementation that serializes information information (the implementation is offered on GitHub)
  • REGISTRY_NAME – The Schema Registry from Account B
  • SCHEMA_NAME – The schema identify from Account B
  • AVRO_RECORD_TYPE – AvroRecordType.SPECIFIC_RECORD

non-public Properties getProducerConfig() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
        props.put(ProducerConfig.ACKS_CONFIG, "-1");
        props.put(ProducerConfig.CLIENT_ID_CONFIG,"msk-cross-account-gsr-producer");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, GlueSchemaRegistryKafkaSerializer.class.getName());
        props.put(AWSSchemaRegistryConstants.DATA_FORMAT, DataFormat.AVRO.identify());
        props.put(AWSSchemaRegistryConstants.AWS_REGION,regionName);
        props.put(AWSSchemaRegistryConstants.REGISTRY_NAME, "unicorn-ride-request-registry");
        props.put(AWSSchemaRegistryConstants.SCHEMA_NAME, "unicorn-ride-request-schema-avro");
        props.put(AWSSchemaRegistryConstants.AVRO_RECORD_TYPE, AvroRecordType.SPECIFIC_RECORD.getName());
        return props;
}

• startProducer() assumes the position in Account B to have the ability to join with the Schema Registry in Account B and sends messages to the Kafka matter on the MSK cluster:

public void startProducer() {
        assumeGlueSchemaRegistryRole();
        KafkaProducer<String, UnicornRideRequest> producer = 
		new KafkaProducer<String,UnicornRideRequest>(getProducerConfig());
        int numberOfMessages = Integer.valueOf(str_numOfMessages);
        logger.information("Beginning to ship information...");
        for(int i = 0;i < numberOfMessages;i ++)
        {
            UnicornRideRequest rideRequest = getRecord(i);
            String key = "key-" + i;
            ProducerRecord<String, UnicornRideRequest> document = 
		new ProducerRecord<String, UnicornRideRequest>(matter, key, rideRequest);
            producer.ship(document, new ProducerCallback());
        }
 }

• assumeGlueSchemaRegistryRole() as proven within the following code makes use of AWS STS to imagine the cross-account Schema Registry IAM position in Account B. (For extra info, see Momentary safety credentials in IAM.) The response from stsClient.assumeRole(roleRequest) comprises the non permanent credentials, which embrace accessKeyId, secretAccessKey, and a sessionToken. It then units the non permanent credentials within the system properties. The AWS SDK for Java makes use of these credentials whereas accessing the Schema Registry (by the Schema Registry serializer). For extra info, see Utilizing Credentials.

  public void assumeGlueSchemaRegistryRole() {
          attempt {
  	   Area area = Area.of(regionName);
              if(!Area.areas().comprises(area))
                   throw new RuntimeException("Area : " + regionName + " is invalid.");
              StsClient stsClient = StsClient.builder().area(area).construct();
              AssumeRoleRequest roleRequest = AssumeRoleRequest.builder()
                      .roleArn(this.assumeRoleARN)
                      .roleSessionName("kafka-producer-cross-account-glue-schemaregistry-demo")
  	           .externalId(this.externalId)	
                      .construct();
              AssumeRoleResponse roleResponse = stsClient.assumeRole(roleRequest);
              Credentials myCreds = roleResponse.credentials();
              System.setProperty("aws.accessKeyId", myCreds.accessKeyId());
              System.setProperty("aws.secretAccessKey", myCreds.secretAccessKey());
              System.setProperty("aws.sessionToken", myCreds.sessionToken());
              stsClient.shut();
          } catch (StsException e) {
              logger.error(e.getMessage());
              System.exit(1);
          }
      }

• createUnicornRideRequest() makes use of the Avro schema (unicorn experience request schema) generated lessons to create a SpecificRecord. For this publish, the unicorn experience request attributes values are hard-coded on this methodology. See the next code:

  public UnicornRideRequest getRecord(int requestId){
              /*
               Initialise UnicornRideRequest object of
               class that's generated from AVRO Schema
               */
             UnicornRideRequest rideRequest = UnicornRideRequest.newBuilder()
              .setRequestId(requestId)
              .setPickupAddress("Melbourne, Victoria, Australia")
              .setDestinationAddress("Sydney, NSW, Aus")
              .setRideFare(1200.50F)
              .setRideDuration(120)
              .setPreferredUnicornColor(UnicornPreferredColor.WHITE)
              .setRecommendedUnicorn(RecommendedUnicorn.newBuilder()
                      .setUnicornId(requestId*2)
                      .setColor(unicorn_color.WHITE)
                      .setStarsRating(5).construct())
              .setCustomer(Buyer.newBuilder()
                      .setCustomerAccountNo(1001)
                      .setFirstName("Dummy")
                      .setLastName("Person")
                      .setEmailAddresses(Arrays.asList("demo@instance.com"))
                      .setCustomerAddress("Flinders Road Station")
                      .setModeOfPayment(ModeOfPayment.CARD)
                      .setCustomerRating(5).construct()).construct();
              logger.information(rideRequest.toString());
              return rideRequest;
      }

Configure the Kafka shopper

The MSKClientStack stack created the KafkaConsumerInstance occasion for the Kafka shopper software. You possibly can view all of the situations created by the stack on the Amazon EC2 console.

To configure the Kafka shopper accessing the Schema Registry within the central AWS account, full the next steps:

1. Open a brand new terminal within the Cloud9EC2Bastion AWS Cloud9 atmosphere.
   
2. SSH into the KafkaConsumerInstance EC2 occasion and set the Area as per your requirement:

   ssh -i <keypair PEM file> ec2-user@<KafkaConsumerInstance Non-public IP handle>
   aws configure set area <area>

3. Set the atmosphere variable MSK_CLUSTER_ARN pointing to the MSK cluster’s ARN:

   export MSK_CLUSTER_ARN=$(aws kafka list-clusters |  jq '.ClusterInfoList[] | choose (.ClusterName == "MSKClusterStack") | {ClusterArn} | be part of (" ")' | tr -d ")

Change the .ClusterName worth in case you used a unique identify for the MSK cluster CloudFormation stack. The cluster identify is identical because the stack identify.

4. Set the atmosphere variable BOOTSTRAP_BROKERS pointing to the bootstrap brokers:

   export BOOTSTRAP_BROKERS=$(aws kafka get-bootstrap-brokers --cluster-arn $MSK_CLUSTER_ARN | jq -r .BootstrapBrokerString)

5. Confirm the atmosphere variables:

   echo $MSK_CLUSTER_ARN
   echo $BOOTSTRAP_BROKERS

The MSKClientStack stack copied the Kafka shopper shopper JAR file referred to as kafka-cross-account-gsr-consumer.jar to the KafkaConsumerInstance occasion. It comprises the Kafka shopper shopper that reads messages from the Kafka matter unicorn-ride-request-topic on the MSK cluster and accesses the unicorn-ride-request-schema-avro Avro schema from the unicorn-ride-request-registry registry in Account B. The Kafka shopper code, which we cowl later on this publish, is offered on GitHub.

6. Run the next instructions and confirm kafka-cross-account-gsr-consumer.jar exists:
7. Run the next command to run the Kafka shopper within the KafkaConsumerInstance terminal:

   java -jar kafka-cross-account-gsr-consumer.jar -bs $BOOTSTRAP_BROKERS 
   -rn <Account B IAM position arn that Kafka shopper software must assume> 
   -topic unicorn-ride-request-topic 
   -reg us-east-1 
   -externalid <Account B IAM position exterior Id that you simply used whereas making a CF stack in Account B>

The code has the next parameters:

• -bs – $BOOTSTRAP_BROKERS (the MSK cluster bootstrap brokers)
• -rn – The CrossAccountGlueSchemaRegistryRoleArn worth from the SchemaRegistryStack stack outputs in Account B
• -topic – The Kafka matter unicorn-ride-request-topic
• -reg – us-east-1 (change it in keeping with your Area, it’s used for the AWS STS endpoint and Schema Registry)
• -externalId – The identical exterior ID (for instance, demo10A) that you simply used whereas creating the CloudFormation stack in Account B

The next screenshot reveals the Kafka shopper logs efficiently studying messages from the Kafka matter on the MSK cluster in Account A and accessing the Avro schema unicorn-ride-request-schema-avro from the unicorn-ride-request-registry schema registry in Account B.

If you happen to see the same logs, it means each the Kafka shopper purposes have been in a position to join efficiently with the centralized Schema Registry in Account B and are in a position to validate messages whereas sending and consuming messages from the MSK cluster in Account A.

Kafka shopper code

The whole Kafka shopper implementation is offered on GitHub. On this part, we break down the code.

• getConsumerConfig() initializes shopper properties, as proven within the following code:
  • VALUE_DESERIALIZER_CLASS_CONFIG – The GlueSchemaRegistryKafkaDeserializer.class.getName() AWS deserializer implementation that deserializes the SpecificRecord as per the encoded schema ID from the Schema Registry (the implementation is offered on GitHub).
  • AVRO_RECORD_TYPE – AvroRecordType.SPECIFIC_RECORD

non-public Properties getConsumerConfig() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "unicorn.riderequest.shopper");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, GlueSchemaRegistryKafkaDeserializer.class.getName());
        props.put(AWSSchemaRegistryConstants.AWS_REGION, regionName);
        props.put(AWSSchemaRegistryConstants.AVRO_RECORD_TYPE, AvroRecordType.SPECIFIC_RECORD.getName());
        return props;
}

• startConsumer() assumes the position in Account B to have the ability to join with the Schema Registry in Account B and reads messages from the Kafka matter on the MSK cluster:

public void startConsumer() {
  logger.information("beginning shopper...");
  assumeGlueSchemaRegistryRole();
  KafkaConsumer<String, UnicornRideRequest> shopper = new KafkaConsumer<String, UnicornRideRequest>(getConsumerConfig());
  shopper.subscribe(Collections.singletonList(matter));
  int rely = 0;
  whereas (true) {
            ultimate ConsumerRecords<String, UnicornRideRequest> information = shopper.ballot(Period.ofMillis(1000));
            for (ultimate ConsumerRecord<String, UnicornRideRequest> document : information) {
                ultimate UnicornRideRequest rideRequest = document.worth();
                logger.information(String.valueOf(rideRequest.getRequestId()));
                logger.information(rideRequest.toString());
            }
        }
}

• assumeGlueSchemaRegistryRole() as proven within the following code makes use of AWS STS to imagine the cross-account Schema Registry IAM position in Account B. The response from stsClient.assumeRole(roleRequest) comprises the non permanent credentials, which embrace accessKeyId, secretAccessKey, and a sessionToken. It then units the non permanent credentials within the system properties. The SDK for Java makes use of these credentials whereas accessing the Schema Registry (by the Schema Registry serializer). For extra info, see Utilizing Credentials.

public void assumeGlueSchemaRegistryRole() {
        attempt {
	Area area = Area.of(regionName);
            if(!Area.areas().comprises(area))
                 throw new RuntimeException("Area : " + regionName + " is invalid.");
            StsClient stsClient = StsClient.builder().area(area).construct();
            AssumeRoleRequest roleRequest = AssumeRoleRequest.builder()
                    .roleArn(this.assumeRoleARN)
                    .roleSessionName("kafka-consumer-cross-account-glue-schemaregistry-demo")
                    .externalId(this.externalId)
                    .construct();
            AssumeRoleResponse roleResponse = stsClient.assumeRole(roleRequest);
            Credentials myCreds = roleResponse.credentials();
            System.setProperty("aws.accessKeyId", myCreds.accessKeyId());
            System.setProperty("aws.secretAccessKey", myCreds.secretAccessKey());
            System.setProperty("aws.sessionToken", myCreds.sessionToken());
            stsClient.shut();
        } catch (StsException e) {
            logger.error(e.getMessage());
            System.exit(1);
        }
    }

Compile and generate Avro schema lessons

Like every other a part of constructing and deploying your software, schema compilation and the method of producing Avro schema lessons needs to be included in your CI/CD pipeline. There are a number of methods to generate Avro schema lessons; we use avro-maven-plugin for this publish. The CI/CD course of also can use avro-tools to compile Avro schema to generate lessons. The next code is an instance of how you should utilize avro-tools:

java -jar /path/to/avro-tools-1.10.2.jar compile schema <schema file> <vacation spot>

//compiling unicorn_ride_request.avsc
java -jar avro-tools-1.10.2.jar compile schema unicorn_ride_request.avsc .

Implementation overview

To recap, we begin with defining and registering an Avro schema for the unicorn experience request message within the AWS Glue Schema Registry in Account B, the central information lake account. In Account A, we create an MSK cluster and Kafka producer and shopper EC2 situations with their respective software code (kafka-cross-account-gsr-consumer.jar and kafka-cross-account-gsr-producer.jar) and deployed in them utilizing the CloudFormation stack.

Once we run the producer software in Account A, the serializer (GlueSchemaRegistryKafkaSerializer) from the AWS Glue Schema Registry SerDe library offered because the configuration will get the unicorn experience request schema (UnicornRideRequest.avsc) from the central Schema Registry residing in Account B to serialize the unicorn experience request message. It makes use of the IAM position (non permanent credentials) in Account B and Area, schema registry identify (unicorn-ride-request-registry), and schema identify (unicorn-ride-request-schema-avro) offered because the configuration to connect with the central Schema Registry. After the message is efficiently serialized, the producer software sends it to the Kafka matter (unicorn-ride-request-topic) on the MSK cluster.

Once we run the buyer software in Account A, the deserializer (GlueSchemaRegistryKafkaDeserializer) from the Schema Registry SerDe library offered because the configuration extracts the encoded schema ID from the message learn from the Kafka matter (unicorn-ride-request-topic) and will get the schema for a similar ID from the central Schema Registry in Account B. It then deserializes the message. It makes use of the IAM position (non permanent credentials) in Account B and the Area offered because the configuration to connect with the central Schema Registry. The buyer software additionally configures Avro’s SPECIFIC_RECORD to tell the deserializer that the message is of a particular kind (unicorn experience request). After the message is efficiently deserialized, the buyer software processes it as per the necessities.

Clear up

The ultimate step is to wash up. To keep away from pointless costs, it’s best to take away all of the sources created by the CloudFormation stacks used for this publish. The only method to take action is to delete the stacks. First delete the MSKClusterStack adopted by MSKClientStack from Account A. Then delete the SchemaRegistryStack from Account B.

Conclusion

On this publish, we demonstrated methods to use AWS Glue Schema Registry with Amazon MSK and stream processing purposes to validate messages utilizing an Avro schema. We created a distributed structure the place the Schema Registry resides in a central AWS account (information lake account) and Kafka producer and shopper purposes reside in a separate AWS account. We created an Avro schema within the schema registry within the central account to make it environment friendly for the applying groups to take care of schemas in a single place. As a result of AWS Glue Schema Registry helps identity-based entry insurance policies, we used the cross-account IAM position to permit the Kafka producer and shopper purposes operating in a separate account to securely entry the schema from the central account to validate messages. As a result of the Avro schema was agreed prematurely, we used Avro SpecificRecord to make sure kind security at compile time and keep away from runtime schema validation points on the shopper aspect. The code used for this publish is offered on GitHub for reference.

To be taught extra concerning the companies and sources on this answer, check with AWS Glue Schema Registry, the Amazon MSK Developer Information, the AWS Glue Schema Registry SerDe library, and IAM tutorial: Delegate entry throughout AWS accounts utilizing IAM roles.

In regards to the Creator

Vikas Bajaj is a Principal Options Architect at Amazon Net Service. Vikas works with digital native clients and advises them on know-how structure and modeling, and choices and options to satisfy strategic enterprise targets. He makes positive designs and options are environment friendly, sustainable, and fit-for-purpose for present and future enterprise wants. Other than structure and know-how discussions, he enjoys watching and enjoying cricket.