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

At the moment’s companies face an unprecedented progress within the quantity of information. A rising portion of the information is generated in actual time by IoT gadgets, web sites, enterprise functions, and varied different sources. Companies have to course of and analyze this knowledge as quickly because it arrives to make enterprise selections in actual time. Amazon Managed Streaming for Apache Kafka (Amazon MSK) is a completely managed service that permits constructing and working stream processing functions that use Apache Kafka to gather and course of knowledge in actual time.

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

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

This put up demonstrates how Apache Kafka stream processing functions 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 functions whereas sending and receiving messages from a Kafka matter on an Amazon MSK cluster. Though we use an Avro schema for this put up, the identical method and idea applies to JSON schemas as properly.

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 trip request messages. They anticipate rides to be extremely popular and need to make sure that their answer can scale. They’re additionally constructing a central knowledge lake the place all their streaming and operation knowledge is saved for evaluation. They’re buyer obsessed, so that they anticipate so as to add new enjoyable options to future rides, like selecting the hair colour of your unicorn, and might want to replicate these attributes within the trip request messages. To keep away from points in downstream functions as a result of future schema adjustments, 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.

Answer overview

The corporate makes use of Amazon MSK to seize and distribute the unicorn trip request messages at scale. They outline an Avro schema for unicorn trip requests as a result of it supplies wealthy knowledge buildings, helps direct mapping to JSON, in addition to a compact, quick, and binary knowledge 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 using an Avro file as a POJO. That is accomplished by producing a Java class (or courses) from the schema, by utilizing avro-maven-plugin. They use AWS Identification and Entry Administration (IAM) cross-account roles to permit producer and shopper functions from the opposite AWS account to securely 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 functions 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 task in Account B utilizing an identity-based coverage as a result of the AWS Glue Schema Registry doesn’t assist resource-based insurance policies.

• Account A IAM function – Permits producer and shopper functions to imagine an IAM function in Account B.
• Account B IAM function – 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 situation, IAM principals that may assume cross-account roles must be scoped extra particularly.

The next structure diagram illustrates the answer:

The answer works as follows:

1. A Kafka producer working in Account A assumes the cross-account Schema Registry IAM function in Account B by calling the AWS Safety Token Service (AWS STS) assumeRole API.
2. The Kafka producer retrieves the unicorn trip request Avro schema model ID from the AWS Glue Schema Registry for the schema that’s embedded within the unicorn trip 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 information file 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 working in Account A assumes the cross-account Schema Registry IAM function in Account B by calling the AWS STS assumeRole API.
5. The Kafka shopper begins polling the Kafka matter on the MSK cluster for knowledge data.
6. The Kafka shopper retrieves the unicorn trip request Avro schema from the AWS Glue Schema Registry, matching the schema model ID that’s encoded within the unicorn trip request knowledge file. 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 information file into the unicorn trip request POJO for the buyer to course of it.

The AWS Glue Schema Registry SerDe library additionally helps optionally available compression configuration to save lots of on knowledge transfers. For extra details about the Schema Registry, see How the Schema Registry works.

Unicorn trip request Avro schema

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

{
    "sort": "file",
    "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",
        "sort": {
          "sort": "file",
          "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", "sort": ["null", "int"], "default": null, "doc": "unicorn star scores primarily based on prospects suggestions"}
          ]
        }
      },
      {
        "identify": "buyer",
        "sort": {
          "sort": "file",
          "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","sort": "string"},
            {"identify": "email_addresses","sort": ["null", {"type":"array", "items":"string"}]},
            {"identify": "customer_address","sort": "string","doc": "buyer deal with"},
            {"identify": "mode_of_payment","sort": {"sort": "enum","identify": "ModeOfPayment","symbols": ["CARD","CASH"]}, "default": "CARD"},
            {"identify": "customer_rating", "sort": ["null", "int"], "default": null}
          ]
        }
      }
    ]
  }

Stipulations

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

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

For this answer, we use Area us-east-1, however you possibly can change this as per your necessities.

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

Create assets in Account B

We create the next assets in Account B:

• A schema registry
• An Avro schema
• An IAM function 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 assets.

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 consumer’s AWS account ID, enter your Account A ID.
7. For ExternalId, enter a singular random ID (for instance, demo10A), which must be supplied by the Kafka shoppers in Account Some time assuming the IAM function on this account.

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

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

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

9. To confirm the assets 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 function 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 function.

12. On the SchemaRegistryStack stack Outputs tab, copy the worth for CrossAccountGlueSchemaRegistryRoleName.
13. On the IAM console, seek for this function.
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 supplied throughout stack creation.

Create assets in Account A

We create the next assets in Account A:

• A VPC
• EC2 cases 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 cases. Additionally create an MSK cluster configuration with default values. You might want to have permissions to create the CloudFormation stack, EC2 cases, AWS Cloud9 atmosphere, MSK cluster, MSK cluster configuration, and IAM function.

1. Log in to Account A.
2. Launch the next CloudFormation stack to launch the VPC, EC2 cases, and AWS Cloud9 atmosphere.
3. For Stack identify, enter MSKClientStack.
4. Present the VPC and subnet CIDR ranges.
5. For EC2 Keypair, select an current EC2 keypair.
6. For the newest EC2 AMI ID, choose the default choice.
7. For the cross-account IAM function ARN, use the worth for CrossAccountGlueSchemaRegistryRoleArn (out there on the Outputs tab of SchemaRegistryStack).
8. Look ahead to 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 based on your model.
14. For the consumer 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 Information.
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 deal with>
   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 a part of (" ")' | tr -d ")

Change the .ClusterName worth within the code for those who used a special 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 known as unicorn-ride-request-topic in your MSK cluster, which is utilized by the Kafka producer and shopper functions 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 consumer JAR file known as kafka-cross-account-gsr-producer.jar to the KafkaProducerInstance occasion. It comprises the Kafka producer consumer 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 put up, is accessible 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 function arn that Kafka producer software must assume> 
   -topic unicorn-ride-request-topic 
   -reg us-east-1 
   -nm 500 
   -externalid <Account B IAM function 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 based on 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 implies 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 accessible 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 knowledge data (the implementation is accessible 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 function 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.data("Beginning to ship data...");
        for(int i = 0;i < numberOfMessages;i ++)
        {
            UnicornRideRequest rideRequest = getRecord(i);
            String key = "key-" + i;
            ProducerRecord<String, UnicornRideRequest> file = 
		new ProducerRecord<String, UnicornRideRequest>(matter, key, rideRequest);
            producer.ship(file, new ProducerCallback());
        }
 }

• assumeGlueSchemaRegistryRole() as proven within the following code makes use of AWS STS to imagine the cross-account Schema Registry IAM function in Account B. (For extra info, see Non permanent safety credentials in IAM.) The response from stsClient.assumeRole(roleRequest) comprises the momentary credentials, which embrace accessKeyId, secretAccessKey, and a sessionToken. It then units the momentary credentials within the system properties. The AWS SDK for Java makes use of these credentials whereas accessing the Schema Registry (by way of 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 trip request schema) generated courses to create a SpecificRecord. For this put up, the unicorn trip request attributes values are hard-coded on this technique. 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("Consumer")
                      .setEmailAddresses(Arrays.asList("demo@instance.com"))
                      .setCustomerAddress("Flinders Avenue Station")
                      .setModeOfPayment(ModeOfPayment.CARD)
                      .setCustomerRating(5).construct()).construct();
              logger.data(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 cases 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 deal with>
   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 a part of (" ")' | tr -d ")

Change the .ClusterName worth for those who used a special 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 consumer JAR file known as kafka-cross-account-gsr-consumer.jar to the KafkaConsumerInstance occasion. It comprises the Kafka shopper consumer 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 put up, is accessible 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 function arn that Kafka shopper software must assume> 
   -topic unicorn-ride-request-topic 
   -reg us-east-1 
   -externalid <Account B IAM function 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 based on 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.

When you see the same logs, it means each the Kafka shopper functions 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 accessible 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 accessible 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 function 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.data("beginning shopper...");
  assumeGlueSchemaRegistryRole();
  KafkaConsumer<String, UnicornRideRequest> shopper = new KafkaConsumer<String, UnicornRideRequest>(getConsumerConfig());
  shopper.subscribe(Collections.singletonList(matter));
  int rely = 0;
  whereas (true) {
            closing ConsumerRecords<String, UnicornRideRequest> data = shopper.ballot(Period.ofMillis(1000));
            for (closing ConsumerRecord<String, UnicornRideRequest> file : data) {
                closing UnicornRideRequest rideRequest = file.worth();
                logger.data(String.valueOf(rideRequest.getRequestId()));
                logger.data(rideRequest.toString());
            }
        }
}

• assumeGlueSchemaRegistryRole() as proven within the following code makes use of AWS STS to imagine the cross-account Schema Registry IAM function in Account B. The response from stsClient.assumeRole(roleRequest) comprises the momentary credentials, which embrace accessKeyId, secretAccessKey, and a sessionToken. It then units the momentary credentials within the system properties. The SDK for Java makes use of these credentials whereas accessing the Schema Registry (by way of 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 courses

Like another a part of constructing and deploying your software, schema compilation and the method of producing Avro schema courses must be included in your CI/CD pipeline. There are a number of methods to generate Avro schema courses; we use avro-maven-plugin for this put up. The CI/CD course of may also use avro-tools to compile Avro schema to generate courses. The next code is an instance of how you should use 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 trip request message within the AWS Glue Schema Registry in Account B, the central knowledge lake account. In Account A, we create an MSK cluster and Kafka producer and shopper EC2 cases 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.

After we run the producer software in Account A, the serializer (GlueSchemaRegistryKafkaSerializer) from the AWS Glue Schema Registry SerDe library supplied because the configuration will get the unicorn trip request schema (UnicornRideRequest.avsc) from the central Schema Registry residing in Account B to serialize the unicorn trip request message. It makes use of the IAM function (momentary credentials) in Account B and Area, schema registry identify (unicorn-ride-request-registry), and schema identify (unicorn-ride-request-schema-avro) supplied 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.

After we run the buyer software in Account A, the deserializer (GlueSchemaRegistryKafkaDeserializer) from the Schema Registry SerDe library supplied 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 function (momentary credentials) in Account B and the Area supplied 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 sort (unicorn trip 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 fees, you need to take away all of the assets created by the CloudFormation stacks used for this put up. The best approach 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 put up, we demonstrated tips on how to use AWS Glue Schema Registry with Amazon MSK and stream processing functions to validate messages utilizing an Avro schema. We created a distributed structure the place the Schema Registry resides in a central AWS account (knowledge lake account) and Kafka producer and shopper functions 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 function to permit the Kafka producer and shopper functions working 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 sort security at compile time and keep away from runtime schema validation points on the consumer facet. The code used for this put up is accessible on GitHub for reference.

To be taught extra concerning the providers and assets on this answer, seek advice from 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 Writer

Vikas Bajaj is a Principal Options Architect at Amazon Internet Service. Vikas works with digital native prospects and advises them on expertise structure and modeling, and choices and options to satisfy strategic enterprise goals. He makes positive designs and options are environment friendly, sustainable, and fit-for-purpose for present and future enterprise wants. Aside from structure and expertise discussions, he enjoys watching and enjoying cricket.