Again up and restore Kafka matter information utilizing Amazon MSK Join

You should use Apache Kafka to run your streaming workloads. Kafka supplies resiliency to failures and protects your information out of the field by replicating information throughout the brokers of the cluster. This makes certain that the information within the cluster is sturdy. You’ll be able to obtain your sturdiness SLAs by altering the replication issue of the subject. Nevertheless, streaming information saved in Kafka matters tends to be transient and usually has a retention time of days or even weeks. You could need to again up the information saved in your Kafka matter lengthy after its retention time expires for a number of causes. For instance, you may need compliance necessities that require you to retailer the information for a number of years. Or you’ll have curated artificial information that must be repeatedly hydrated into Kafka matters earlier than beginning your workload’s integration checks. Or an upstream system that you just don’t have management over produces unhealthy information and it’s good to restore your matter to a beforehand nicely state.

Storing information indefinitely in Kafka matters is an possibility, however typically the use case requires a separate copy. Instruments corresponding to MirrorMaker allow you to again up your information into one other Kafka cluster. Nevertheless, this requires one other energetic Kafka cluster to be operating as a backup, which will increase compute prices and storage prices. A cheap and sturdy means of backing up the information of your Kafka cluster is to make use of an object storage service like Amazon Easy Storage Service (Amazon S3).

On this submit, we stroll by an answer that permits you to again up your information for chilly storage utilizing Amazon MSK Join. We restore the backed-up information to a different Kafka matter and reset the buyer offsets based mostly in your use case.

Overview of answer

Kafka Join is a element of Apache Kafka that simplifies streaming information between Kafka matters and exterior programs like object shops, databases, and file programs. It makes use of sink connectors to stream information from Kafka matters to exterior programs, and supply connectors to stream information from exterior programs to Kafka matters. You should use off-the-shelf connectors written by third events or write your individual connectors to satisfy your particular necessities.

MSK Join is a characteristic of Amazon Managed Streaming for Apache Kafka (Amazon MSK) that permits you to run absolutely managed Kafka Join workloads. It really works with MSK clusters and with appropriate self-managed Kafka clusters. On this submit, we use the Lenses AWS S3 Connector to again up the information saved in a subject in an Amazon MSK cluster to Amazon S3 and restore this information again to a different matter. The next diagram reveals our answer structure.

To implement this answer, we full the next steps:

1. Again up the information utilizing an MSK Join sink connector to an S3 bucket.
2. Restore the information utilizing an MSK Join supply connector to a brand new Kafka matter.
3. Reset shopper offsets based mostly on completely different eventualities.

Conditions

Ensure to finish the next steps as conditions:

1. Arrange the required assets for Amazon MSK, Amazon S3, and AWS Id and Entry Administration (IAM).
2. Create two Kafka matters within the MSK cluster: source_topic and target_topic.
3. Create an MSK Join plugin utilizing the Lenses AWS S3 Connector.
4. Set up the Kafka CLI by following Step 1 of Apache Kafka Quickstart.
5. Set up the kcat utility to ship check messages to the Kafka matter.

Again up your matters

Relying on the use case, chances are you’ll need to again up all of the matters in your Kafka cluster or again up some particular matters. On this submit, we cowl methods to again up a single matter, however you’ll be able to lengthen the answer to again up a number of matters.

The format during which the information is saved in Amazon S3 is essential. You could need to examine the information that’s saved in Amazon S3 to debug points just like the introduction of unhealthy information. You’ll be able to look at information saved as JSON or plain textual content through the use of textual content editors and searching within the time frames which can be of curiosity to you. You may as well look at massive quantities of knowledge saved in Amazon S3 as JSON or Parquet utilizing AWS companies like Amazon Athena. The Lenses AWS S3 Connector helps storing objects as JSON, Avro, Parquet, plaintext, or binary.

On this submit, we ship JSON information to the Kafka matter and retailer it in Amazon S3. Relying on the information sort that meets your necessities, replace the join.s3.kcql assertion and *.converter configuration. You’ll be able to consult with the Lenses sink connector documentation for particulars of the codecs supported and the associated configurations. If the prevailing connectors don’t work on your use case, you can even write your individual connector or lengthen present connectors. You’ll be able to partition the information saved in Amazon S3 based mostly on fields of primitive sorts within the message header or payload. We use the date fields saved within the header to partition the information on Amazon S3.

Observe these steps to again up your matter:

1. Create a brand new Amazon MSK sink connector by operating the next command:

   aws kafkaconnect create-connector 
   --capacity "autoScaling={maxWorkerCount=2,mcuCount=1,minWorkerCount=1,scaleInPolicy={cpuUtilizationPercentage=10},scaleOutPolicy={cpuUtilizationPercentage=80}}" 
   --connector-configuration 
   "connector.class=io.lenses.streamreactor.join.aws.s3.sink.S3SinkConnector, 
   key.converter.schemas.allow=false, 
   join.s3.kcql=INSERT INTO <<S3 Bucket Identify>>:my_workload SELECT * FROM source_topic PARTITIONBY _header.12 months,_header.month,_header.day,_header.hour STOREAS `JSON` WITHPARTITIONER=KeysAndValues WITH_FLUSH_COUNT = 5, 
   aws.area=us-east-1, 
   duties.max=2, 
   matters=source_topic, 
   schema.allow=false, 
   errors.log.allow=true, 
   worth.converter=org.apache.kafka.join.storage.StringConverter, 
   key.converter=org.apache.kafka.join.storage.StringConverter " 
   --connector-name "backup-msk-to-s3-v1" 
   --kafka-cluster '{"apacheKafkaCluster": {"bootstrapServers": "<<MSK dealer record>>","vpc": {"securityGroups": [ <<Security Group>> ],"subnets": [ <<Subnet List>> ]}}}' 
   --kafka-cluster-client-authentication "authenticationType=NONE" 
   --kafka-cluster-encryption-in-transit "encryptionType=PLAINTEXT" 
   --kafka-connect-version "2.7.1" 
   --plugins "customPlugin={customPluginArn=<< ARN of the MSK Join Plugin >>,revision=1}" 
   --service-execution-role-arn " <<ARN of the IAM Position>> "

2. Ship information to the subject utilizing kcat:

   ./kcat -b <<dealer record>> -t source_topic -H "12 months=$(date +"%Y")" -H "month=$(date +"%m")" -H "day=$(date +"%d")" -H "hour=$(date +"%H")" -P
   {"message":"interesset eros vel elit salutatus"}
   {"message":"impetus deterruisset per aliquam luctus"}
   {"message":"ridens vocibus feugait vitae cras"}
   {"message":"interesset eros vel elit salutatus"}
   {"message":"impetus deterruisset per aliquam luctus"}
   {"message":"ridens vocibus feugait vitae cras"}

3. Test the S3 bucket to verify the information is being written.

MSK Join publishes metrics to Amazon CloudWatch that you should use to observe your backup course of. Necessary metrics are SinkRecordReadRate and SinkRecordSendRate, which measure the typical variety of information learn from Kafka and written to Amazon S3, respectively.

Additionally, ensure that the backup connector is maintaining with the speed at which the Kafka matter is receiving messages by monitoring the offset lag of the connector. In case you’re utilizing Amazon MSK, you are able to do this by turning on partition-level metrics on Amazon MSK and monitoring the OffsetLag metric of all of the partitions for the backup connector’s shopper group. It’s best to hold this as near 0 as attainable by adjusting the utmost variety of MSK Join employee cases. The command that we used within the earlier step units MSK Connect with routinely scale as much as two staff. Alter the --capacity setting to extend or lower the utmost employee depend of MSK Join staff based mostly on the OffsetLag metric.

Restore information to your matters

You’ll be able to restore your backed-up information to a brand new matter with the identical title in the identical Kafka cluster, a distinct matter in the identical Kafka cluster, or a distinct matter in a distinct Kafka cluster altogether. On this submit, we stroll by the situation of restoring information that was backed up in Amazon S3 to a distinct matter, target_topic, in the identical Kafka cluster. You’ll be able to lengthen this to different eventualities by altering the subject and dealer particulars within the connector configuration.

Observe these steps to revive the information:

1. Create an Amazon MSK supply connector by operating the next command:

   aws kafkaconnect create-connector 
   --capacity "autoScaling={maxWorkerCount=2,mcuCount=1,minWorkerCount=1,scaleInPolicy={cpuUtilizationPercentage=10},scaleOutPolicy={cpuUtilizationPercentage=80}}"   
   --connector-configuration 
       "connector.class=io.lenses.streamreactor.join.aws.s3.supply.S3SourceConnector, 
        key.converter.schemas.allow=false, 
        join.s3.kcql=INSERT INTO target_topic SELECT * FROM <<S3 Bucket Identify>>:my_workload PARTITIONBY _header.12 months,_header.month,_header.day,_header.hour STOREAS `JSON` WITHPARTITIONER=KeysAndValues WITH_FLUSH_COUNT = 5 , 
        aws.area=us-east-1, 
        duties.max=2, 
        matters=target_topic, 
        schema.allow=false, 
        errors.log.allow=true, 
        worth.converter=org.apache.kafka.join.storage.StringConverter, 
        key.converter=org.apache.kafka.join.storage.StringConverter " 
   --connector-name "restore-s3-to-msk-v1" 
   --kafka-cluster '{"apacheKafkaCluster": {"bootstrapServers": "<<MSK dealer record>>","vpc": {"securityGroups": [<<Security Group>>],"subnets": [ <<Subnet List>> ]}}}' 
   --kafka-cluster-client-authentication "authenticationType=NONE" 
   --kafka-cluster-encryption-in-transit "encryptionType=PLAINTEXT" 
   --kafka-connect-version "2.7.1" 
   --plugins "customPlugin={customPluginArn=<< ARN of the MSK Join Plugin >>,revision=1}" 
   --service-execution-role-arn " <<ARN of the IAM Position>> "

The connector reads the information from the S3 bucket and replays it again to target_topic.

2. Confirm if the information is being written to the Kafka matter by operating the next command:

   ./kafka-console-consumer.sh --bootstrap-server <<MSK dealer record>> --topic target_topic --from-beginning

MSK Join connectors run indefinitely, ready for brand spanking new information to be written to the supply. Nevertheless, whereas restoring, it’s important to cease the connector in any case the information is copied to the subject. MSK Join publishes the SourceRecordPollRate and SourceRecordWriteRate metrics to CloudWatch, which measure the typical variety of information polled from Amazon S3 and variety of information written to the Kafka cluster, respectively. You’ll be able to monitor these metrics to trace the standing of the restore course of. When these metrics attain 0, the information from Amazon S3 is restored to the target_topic. You may get notified of the completion by organising a CloudWatch alarm on these metrics. You’ll be able to lengthen the automation to invoke an AWS Lambda perform that deletes the connector when the restore is full.

As with the backup course of, you’ll be able to velocity up the restore course of by scaling out the variety of MSK Join staff. Change the --capacity parameter to regulate the utmost and minimal staff to a quantity that meets the restore SLAs of your workload.

Reset shopper offsets

Relying on the necessities of restoring the information to a brand new Kafka matter, you may additionally must reset the offsets of the shopper group earlier than consuming or producing to them. Figuring out the precise offset that you just need to reset to is dependent upon your particular enterprise use case and entails guide work to establish this. You should use instruments like Amazon S3 Choose, Athena, or different customized instruments to examine the objects. The next screenshot demonstrates studying the information ending at offset 14 of partition 2 of matter source_topic utilizing S3 Choose.

After you establish the brand new begin offsets on your shopper teams, it’s important to reset them in your Kafka cluster. You are able to do this utilizing the CLI instruments that come bundled with Kafka.

Current shopper teams

If you wish to use the identical shopper group title after restoring the subject, you are able to do this by operating the next command for every partition of the restored matter:

 ./kafka-consumer-groups.sh --bootstrap-server <<dealer record>> --group <<shopper group>> --topic target_topic:<<partition>> --to-offset <<desired offset>> --reset-offsets --execute

Confirm this by operating the --describe possibility of the command:

./kafka-consumer-groups.sh --bootstrap-server <<dealer record>> --group <<shopper group>>  --describe
TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        ...
source_topic  0          211006          188417765       188206759  ...
source_topic  1          212847          192997707       192784860  ...
source_topic  2          211147          196410627       196199480  ...
target_topic  0          211006          188417765       188206759  ...
target_topic  1          212847          192997707       192784860  ...
target_topic  2          211147          196410627       196199480  ...

New shopper group

If you need your workload to create a brand new shopper group and search to customized offsets, you are able to do this by invoking the search methodology in your Kafka shopper for every partition. Alternatively, you’ll be able to create the brand new shopper group by operating the next code:

./kafka-console-consumer.sh --bootstrap-server <<dealer record>> --topic target_topic --group <<shopper group>> --from-beginning --max-messages 1

Reset the offset to the specified offsets for every partition by operating the next command:

./kafka-consumer-groups.sh --bootstrap-server <<dealer record>> --group <<New shopper group>> --topic target_topic:<<partition>> --to-offset <<desired offset>> --reset-offsets --execute

Clear up

To keep away from incurring ongoing expenses, full the next cleanup steps:

1. Delete the MSK Join connectors and plugin.
2. Delete the MSK cluster.
3. Delete the S3 buckets.
4. Delete any CloudWatch assets you created.

Conclusion

On this submit, we confirmed you methods to again up and restore Kafka matter information utilizing MSK Join. You’ll be able to lengthen this answer to a number of matters and different information codecs based mostly in your workload. Make sure to check numerous eventualities that your workloads could face and doc the runbook for every of these eventualities.

For extra info, see the next assets:

In regards to the Writer

Rakshith Rao is a Senior Options Architect at AWS. He works with AWS’s strategic clients to construct and function their key workloads on AWS.