Parameters:
  S3BucketName:
    Type: String
    Description: An Amazon S3 bucket that is used to store data, job scripts, etc.

  DatabaseUserName:
    Type: String
    Description: A user name for RDS database instance.
    NoEcho: true
    MinLength: 1
    MaxLength: 16
    AllowedPattern: '[a-zA-Z][a-zA-Z0-9]*'    
    ConstraintDescription: must begin with a letter and contain only alphanumeric characters.

  DatabaseUserPassword:
    Type: String
    Description: A user password for RDS database instance. (8 characters minimum, 41 characters maximum.)
    NoEcho: true
    MinLength: 8
    MaxLength: 41
    AllowedPattern: '[a-zA-Z0-9]*'
    ConstraintDescription: must contain only alphanumeric characters.

  VPCCIDR:
    Type: String
    Description: CIDR of VPC. IPv4 address range in CIDR notation.
    Default: "10.1.0.0/16"

  PublicSubnetCIDR:
    Type: String
    Description: CIDR of a public subnet A. IPv4 address range in CIDR notation.
    Default: "10.1.0.0/24"

  PrivateSubnetACIDR:
    Type: String
    Description: CIDR of a private subnet A. IPv4 address range in CIDR notation.
    Default: "10.1.100.0/24"

  PrivateSubnetBCIDR:
    Type: String
    Description: CIDR of a private subnet B. IPv4 address range in CIDR notation.
    Default: "10.1.200.0/24"

  SubnetAzA:
    Type: AWS::EC2::AvailabilityZone::Name
    Description: Availability zone for the private subnet A.

  SubnetAzB:
    Type: AWS::EC2::AvailabilityZone::Name
    Description: Availability zone for the private subnet B.

Metadata:
  AWS::CloudFormation::Interface:
    ParameterGroups:
      -
        Label:
          default: General
        Parameters:
          - S3BucketName
      -
        Label:
          default: Network Configuration
        Parameters:
          - VPCCIDR
          - PublicSubnetCIDR
          - PrivateSubnetACIDR
          - PrivateSubnetBCIDR
          - SubnetAzA
          - SubnetAzB

Resources:

  S3Bucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Ref S3BucketName
      AccessControl: Private
      PublicAccessBlockConfiguration:
        BlockPublicAcls: True
        BlockPublicPolicy: True
        IgnorePublicAcls: True
        RestrictPublicBuckets: True

  GlueServiceRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - glue.amazonaws.com
            Action:
              - sts:AssumeRole
      RoleName: !Sub GlueServiceRole-${AWS::StackName}
      Path: /
      Policies:
        - PolicyDocument:
            Version: '2012-10-17'
            Statement:
              -
                Effect: "Allow"
                Action:
                  - s3:GetObject
                  - s3:PutObject
                  - s3:DeleteObject
                Resource:
                  - !Sub arn:aws:s3:::${S3BucketName}/*
              -
                Effect: "Allow"
                Action:
                  - s3:ListBucket
                  - s3:GetLifecycleConfiguration
                Resource:
                  - !Sub arn:aws:s3:::${S3BucketName}
          PolicyName: !Sub GlueServiceRole-S3-${AWS::StackName}
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
        - arn:aws:iam::aws:policy/AmazonKinesisReadOnlyAccess
        - arn:aws:iam::aws:policy/SecretsManagerReadWrite
        


  GlueStreamingJob:
    Type: AWS::Glue::Job
    DependsOn:      
      - KinesisDataStreamCdc
    Properties:
      Name: !Sub 'streaming-cdc-kinesis2hudi-${AWS::StackName}'      
      Command:
        Name: gluestreaming
        ScriptLocation: !Sub 's3://${S3BucketName}/script/streaming-cdc-kinesis2hudi.py'
        PythonVersion: 3
      DefaultArguments:
        --TempDir: !Sub 's3://${S3BucketName}/tmp/'
        --job-language: python
        --job-bookmark-option: job-bookmark-disable
        --enable-metrics: ''
        --enable-continuous-cloudwatch-log: true        
        --enable-glue-datacatalog: "true"
        --database_name: !Ref GlueDatabase
        --kinesis_table_name: !Ref GlueTableKinesis
        --s3_path_hudi_table: !Sub 's3://${S3BucketName}/retail/'
        --s3_path_spark: !Sub 's3://${S3BucketName}/spark_checkpoints/'                
        --starting_position_of_kinesis_iterator: TRIM_HORIZON
        --window_size: '60 seconds' 
        --datalake-formats: hudi
        
      WorkerType: G.1X
      NumberOfWorkers: 2
      GlueVersion: 4.0
      MaxRetries: 0
      Role: !GetAtt GlueServiceRole.Arn
      ExecutionProperty:
        MaxConcurrentRuns: 1




  GlueInitialRDSSetupJob:
    DependsOn:
      - RDS
      - RDSSecret
    Type: AWS::Glue::Job
    Properties:
      Name: !Sub 'rds-ingest-rds-setup-${AWS::StackName}'
      Role: !Ref GlueServiceRole
      Command:
        Name: pythonshell
        ScriptLocation: !Sub 's3://${S3BucketName}/script/rds_data_initial_setup.py'
        PythonVersion: 3.9
      DefaultArguments:
        --mode: ddl
        --secret_name: !Sub "${AWS::StackName}-my-secret"
        --TempDir: !Sub 's3://${S3BucketName}/tmp/'
        --job-language: python
        --additional-python-modules: mysql-connector-python==8.0.32,Faker==8.11.0
      GlueVersion: 3.0
      MaxCapacity: 0.0625
      Connections:
        Connections:
          - !Ref GlueMySQLConnection


  GlueInitialDataLoadRDSJob:
    DependsOn:
      - RDS
      - RDSSecret
    Type: AWS::Glue::Job
    Properties:
      Name: !Sub 'rds-ingest-data-initial-${AWS::StackName}'
      Role: !Ref GlueServiceRole
      Command:
        Name: pythonshell
        ScriptLocation: !Sub 's3://${S3BucketName}/script/rds_data_initial_ingest.py'
        PythonVersion: 3.9
      DefaultArguments:
        --mode: initial
        --secret_name: !Sub "${AWS::StackName}-my-secret"
        --TempDir: !Sub 's3://${S3BucketName}/tmp/'
        --job-language: python
        --additional-python-modules: mysql-connector-python==8.0.32,Faker==8.11.0
      GlueVersion: 3.0
      MaxCapacity: 0.0625
      Connections:
        Connections:
          - !Ref GlueMySQLConnection

  GlueIncrementalDataLoadRDSJob:
    DependsOn:
      - RDS
      - RDSSecret
    Type: AWS::Glue::Job
    Properties:
      Name: !Sub 'rds-ingest-data-incremental-${AWS::StackName}'
      Role: !Ref GlueServiceRole
      Command:
        Name: pythonshell
        ScriptLocation: !Sub 's3://${S3BucketName}/script/rds_data_initial_ingest.py'
        PythonVersion: 3.9
      DefaultArguments:
        --mode: incremental
        --secret_name: !Sub "${AWS::StackName}-my-secret"
        --TempDir: !Sub 's3://${S3BucketName}/tmp/'
        --job-language: python
        --additional-python-modules: mysql-connector-python==8.0.32,Faker==8.11.0
      ExecutionProperty:
        MaxConcurrentRuns: 3
      GlueVersion: 3.0
      MaxCapacity: 0.0625
      Connections:
        Connections:
          - !Ref GlueMySQLConnection

  GlueUpsertRDSJob:
    DependsOn:
      - RDS
      - RDSSecret
    Type: AWS::Glue::Job
    Properties:
      Name: !Sub 'rds-upsert-data-${AWS::StackName}'
      Role: !Ref GlueServiceRole
      Command:
        Name: pythonshell
        ScriptLocation: !Sub 's3://${S3BucketName}/script/rds_upsert_data.py'
        PythonVersion: 3.9
      DefaultArguments:
        --ticketactivity_id: '1'
        --secret_name: !Sub "${AWS::StackName}-my-secret"
        --TempDir: !Sub 's3://${S3BucketName}/tmp/'
        --job-language: python
        --additional-python-modules: mysql-connector-python==8.0.32,Faker==8.11.0
      GlueVersion: 3.0
      MaxCapacity: 0.0625
      Connections:
        Connections:
          - !Ref GlueMySQLConnection

  PeriodicalTrigger:
    DependsOn:
      - GlueIncrementalDataLoadRDSJob
      - RDS
    Type: AWS::Glue::Trigger
    Properties:
      Name: !Sub 'periodical-trigger-${AWS::StackName}'
      Description: Glue trigger which triggers incremental data load to RDS
      Type: SCHEDULED
      Schedule: cron(* * * * ? *)
      Actions:
        - JobName: !Ref GlueIncrementalDataLoadRDSJob

  GlueMySQLConnection:
    DependsOn:
      - SecurityGroup
      - RDS
    Type: AWS::Glue::Connection
    Properties:
      ConnectionInput:
        Name: !Sub 'mysql-connection-${AWS::StackName}'
        ConnectionType: JDBC
        MatchCriteria: []
        PhysicalConnectionRequirements:
          AvailabilityZone: !Ref SubnetAzA
          SecurityGroupIdList:
            - !GetAtt SecurityGroup.GroupId
          SubnetId: !Ref PrivateSubnetA
        ConnectionProperties:
          JDBC_CONNECTION_URL: !Sub "jdbc:mysql://${RDS.Endpoint.Address}:${RDS.Endpoint.Port}/dev"
          JDBC_ENFORCE_SSL: false
          USERNAME: !Ref DatabaseUserName
          PASSWORD: !Ref DatabaseUserPassword
      CatalogId: !Ref AWS::AccountId


  S3CustomResource:
    DependsOn: S3Bucket
    Type: Custom::S3CustomResource
    Properties:
      ServiceToken: !GetAtt LambdaFunction.Arn
      the_bucket: !Ref S3BucketName
      the_streaming_job_script_file_key: script/streaming-cdc-kinesis2hudi.py
      the_rds_data_initial_ingest_script_file_key: script/rds_data_initial_ingest.py
      the_rds_data_initial_setup_script_file_key: script/rds_data_initial_setup.py
      the_rds_upsert_job_script_file_key: script/rds_upsert_data.py
      the_origin_streaming_job_script_url: s3://aws-blogs-artifacts-public/artifacts/BDB-2503/streaming-cdc-kinesis2hudi.py
      the_origin_rds_data_initial_ingest_url: s3://aws-blogs-artifacts-public/artifacts/BDB-2503/rds_data_initial_ingest.py
      the_origin_rds_data_initial_setup_url: s3://aws-blogs-artifacts-public/artifacts/BDB-2503/rds_data_initial_setup.py
      the_origin_rds_upsert_job_script_url: s3://aws-blogs-artifacts-public/artifacts/BDB-2503/rds_upsert_data.py
      the_origin_mysql_connector_url: https://files.pythonhosted.org/packages/d5/f8/1bd20c81de2487c35b20e08a1e079534a7b579b83a5bc85a4485673b7f79/mysql_connector_python-8.0.25-cp38-cp38-manylinux1_x86_64.whl
      the_origin_faker_url: https://files.pythonhosted.org/packages/b2/7a/42bb20f3b612a7636c484ac2a17988d4a98ad14d169937c668c1962599e7/Faker-8.11.0-py3-none-any.whl

  LambdaFunction:
    Type: "AWS::Lambda::Function"
    Properties:
      FunctionName: !Sub 'lambda-custom-resource-${AWS::StackName}'
      Description: This is used as an AWS CloudFormation custom resource to copy job scripts from an AWS Glue managed GitHub repository and an AWS Big Data blog  S3 bucket to your S3 bucket.
      Handler: index.handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Timeout: 360
      Runtime: python3.8
      Code:
        ZipFile: !Sub
          - |
            import boto3                        
            from botocore.client import ClientError
            import cfnresponse
            from urllib.parse import urlparse
            import urllib.request

            def handler(event, context):
                # Init
                print(f'event: {event}')
                the_event = event['RequestType'].strip()
                print("The event is: ", the_event)
                response_data = {}
                s3 = boto3.client('s3')
                s3_resource = boto3.resource('s3')
                # Retrieve parameters
                the_bucket = event['ResourceProperties']['the_bucket'].strip()
                the_streaming_job_script_file_key = event['ResourceProperties']['the_streaming_job_script_file_key'].strip()
                the_rds_data_initial_ingest_script_file_key = event['ResourceProperties']['the_rds_data_initial_ingest_script_file_key'].strip()
                the_rds_data_initial_setup_script_file_key = event['ResourceProperties']['the_rds_data_initial_setup_script_file_key'].strip()
                the_rds_upsert_job_script_file_key = event['ResourceProperties']['the_rds_upsert_job_script_file_key'].strip()
                the_origin_streaming_job_script_url = event['ResourceProperties']['the_origin_streaming_job_script_url'].strip()                
                the_origin_rds_data_initial_ingest_url = event['ResourceProperties']['the_origin_rds_data_initial_ingest_url'].strip()
                the_origin_rds_data_initial_setup_url = event['ResourceProperties']['the_origin_rds_data_initial_setup_url'].strip()
                the_origin_rds_upsert_job_script_url = event['ResourceProperties']['the_origin_rds_upsert_job_script_url'].strip()
                the_origin_mysql_connector_url = event['ResourceProperties']['the_origin_mysql_connector_url'].strip()
                the_origin_faker_url = event['ResourceProperties']['the_origin_faker_url'].strip()
                try:
                    if the_event in ('Create', 'Update'):
                        # Copying job script
                        try:
                            s3_source_object = urlparse(the_origin_streaming_job_script_url, allow_fragments=False)
                            copy_source = {
                                          'Bucket': s3_source_object.netloc,
                                          'Key': s3_source_object.path.lstrip('/')
                                      }
                            s3_resource.meta.client.copy(copy_source, the_bucket, the_streaming_job_script_file_key)
                            

                            s3_source_object = urlparse(the_origin_rds_data_initial_ingest_url, allow_fragments=False)
                            copy_source = {
                                'Bucket': s3_source_object.netloc,
                                'Key': s3_source_object.path.lstrip('/')
                            }
                            s3_resource.meta.client.copy(copy_source, the_bucket, the_rds_data_initial_ingest_script_file_key)

                            
                            s3_source_object = urlparse(the_origin_rds_data_initial_setup_url, allow_fragments=False)
                            copy_source = {
                                'Bucket': s3_source_object.netloc,
                                'Key': s3_source_object.path.lstrip('/')
                            }
                            s3_resource.meta.client.copy(copy_source, the_bucket, the_rds_data_initial_setup_script_file_key)


                            s3_source_object = urlparse(the_origin_rds_upsert_job_script_url, allow_fragments=False)
                            copy_source = {
                                'Bucket': s3_source_object.netloc,
                                'Key': s3_source_object.path.lstrip('/')
                            }
                            s3_resource.meta.client.copy(copy_source, the_bucket, the_rds_upsert_job_script_file_key)



                            reqW = urllib.request.Request(the_origin_mysql_connector_url, method='GET')
                            urlDataW = urllib.request.urlopen(reqW).read()

                            
                        except ClientError as ce:
                            print("Failed to copy the source code file.")
                            print(ce)
                            print(ce.response['ResponseMetadata'])
                        except urllib.error.HTTPError as e:
                            print(e)
                    elif the_event == 'Delete':
                        try:
                          pages = []
                          paginator = s3.get_paginator('list_objects_v2')
                          for page in paginator.paginate(Bucket=the_bucket):
                            pages.extend(page['Contents'])
                          for source in pages:
                            print('deleting')                            
                            s3.delete_object(Bucket=the_bucket,Key=source["Key"])
                            print(source["Key"])
                        except ClientError as ce:
                            print(f"Failed to delete the files: {ce}")

                    # Everything OK... send the signal back
                    print("Completed.")
                    cfnresponse.send(event,
                                      context,
                                      cfnresponse.SUCCESS,
                                      response_data)
                except Exception as e:
                    print("Failed...")
                    print(str(e))
                    response_data['Data'] = str(e)
                    cfnresponse.send(event,
                                      context,
                                      cfnresponse.FAILED,
                                      response_data)          
          - Region: !Ref AWS::Region

  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub LambdaExecutionRole-${AWS::StackName}
      Description: Runs the Lambda function that has permission to upload the job scripts to the S3 bucket.
      AssumeRolePolicyDocument:
        Statement:
          - Action:
              - sts:AssumeRole
            Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
        Version: '2012-10-17'
      Path: "/"
      Policies:
        - PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Effect: Allow
                Resource: arn:aws:logs:*:*:*
          PolicyName: !Sub AWSLambda-CW-${AWS::StackName}
        - PolicyDocument:
            Version: '2012-10-17'
            Statement:
              -
                Effect: "Allow"
                Action:
                  - s3:PutObject
                  - s3:DeleteObject
                Resource:
                  - !Sub arn:aws:s3:::${S3BucketName}/*
              -
                Effect: "Allow"
                Action:
                  - s3:ListBucket
                  - s3:GetLifecycleConfiguration
                Resource:
                  - !Sub arn:aws:s3:::${S3BucketName}

              -
                Effect: "Allow"
                Action:
                    - s3:GetObjectAcl
                    - s3:GetObject
                    - s3:GetBucketTagging,
                    - s3:GetObjectTagging
                    - s3:GetBucketAcl
                    - s3:GetBucketPolicy
                    - s3:GetObjectVersion
                Resource:
                  - arn:aws:s3:::aws-blogs-artifacts-public/*
                  - arn:aws:s3:::aws-blogs-artifacts-public

          PolicyName: !Sub AWSLambda-S3-${AWS::StackName}

  VPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: !Ref VPCCIDR
      EnableDnsSupport: true
      EnableDnsHostnames: true
      InstanceTenancy: default
      Tags:
        - Key: Name
          Value: !Sub "vpc-${AWS::StackName}"

  PublicSubnet:
    DependsOn: VPC
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone: !Ref SubnetAzA
      CidrBlock: !Ref PublicSubnetCIDR
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: !Sub "public-subnet-${AWS::StackName}"

  PrivateSubnetA:
    DependsOn: VPC
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone: !Ref SubnetAzA
      CidrBlock: !Ref PrivateSubnetACIDR
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: !Sub "private-subnetA-${AWS::StackName}"

  PrivateSubnetB:
    DependsOn: VPC
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone: !Ref SubnetAzB
      CidrBlock: !Ref PrivateSubnetBCIDR
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: !Sub "private-subnetB-${AWS::StackName}"

  PublicRouteTable:
    DependsOn:
      - InternetGatewayAttachment
      - VPC
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: !Sub "public-route-${AWS::StackName}"

  PrivateRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: !Sub "private-route-${AWS::StackName}"

  InternetGatewayRoute:
    Type: AWS::EC2::Route
    Properties:
      RouteTableId: !Ref PublicRouteTable
      DestinationCidrBlock: "0.0.0.0/0"
      GatewayId: !Ref InternetGateway

  NatGatewayRoute:
    Type: AWS::EC2::Route
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      DestinationCidrBlock: "0.0.0.0/0"
      NatGatewayId: !Ref NATGateway

  PublicSubnetRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PublicSubnet
      RouteTableId: !Ref PublicRouteTable

  PrivateSubnetARouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PrivateSubnetA
      RouteTableId: !Ref PrivateRouteTable

  PrivateSubnetBRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PrivateSubnetB
      RouteTableId: !Ref PrivateRouteTable

  SecurityGroup:
    DependsOn: VPC
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: !Sub "security-group-${AWS::StackName}"
      VpcId: !Ref VPC
      SecurityGroupEgress:
      - IpProtocol: "-1"
        CidrIp: 0.0.0.0/0      

  SecurityGroupIngress:
    DependsOn: SecurityGroup
    Type: AWS::EC2::SecurityGroupIngress
    Properties:
      GroupId: !GetAtt SecurityGroup.GroupId
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !GetAtt SecurityGroup.GroupId

  InternetGateway:
    Type: AWS::EC2::InternetGateway
    Properties:
      Tags:
        - Key: Name
          Value: !Sub "igw-${AWS::StackName}"

  InternetGatewayAttachment:
    DependsOn: VPC
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      InternetGatewayId: !Ref InternetGateway
      VpcId: !Ref VPC

  NATGateway:
    DependsOn:
      - NATGatewayEIP
      - PublicSubnet
    Type: AWS::EC2::NatGateway
    Properties:
      AllocationId: !GetAtt NATGatewayEIP.AllocationId
      SubnetId: !Ref PublicSubnet
      Tags:
        - Key: Name
          Value: !Sub "natgw-${AWS::StackName}"

  NATGatewayEIP:
    Type: AWS::EC2::EIP
    Properties:
      Domain: vpc

  RDS:
    DependsOn: SecurityGroup
    Type: AWS::RDS::DBInstance
    Properties:
      DBInstanceIdentifier: !Sub "cdc-sample-${AWS::StackName}"
      DBInstanceClass: db.t3.medium
      Engine: mysql
      EngineVersion: 8.0.32
      StorageType: gp2
      AllocatedStorage: 20
      PubliclyAccessible: false
      StorageEncrypted: true
      DeletionProtection: true
      MultiAZ: false
      DBParameterGroupName: !Ref DBParameterGroup
      DBSubnetGroupName: !Ref DBSubnetGroup
      VPCSecurityGroups:
        - !GetAtt SecurityGroup.GroupId
      DBName: dev
      MasterUsername: !Ref DatabaseUserName
      MasterUserPassword: !Ref DatabaseUserPassword

  DBSubnetGroup:
    Type: AWS::RDS::DBSubnetGroup
    Properties:
      DBSubnetGroupDescription: !Sub "cdc-sample-${AWS::StackName}"
      SubnetIds:
        - !Ref PrivateSubnetA
        - !Ref PrivateSubnetB

  DBParameterGroup:
    Type: AWS::RDS::DBParameterGroup
    Properties:
      Description: !Sub "cdc-sample-${AWS::StackName}"
      Family: mysql8.0
      Parameters:
        general_log: 1
        slow_query_log: 1
        long_query_time: 0
        log_output: FILE
        binlog_format: ROW

  KinesisDataStreamCdc:
    Type: AWS::Kinesis::Stream
    Properties:
      Name: !Sub dms-cdc-${AWS::StackName}
      ShardCount: 2

  DmsKinesisRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - dms.amazonaws.com
            Action:
              - sts:AssumeRole
      RoleName: !Sub DmsKinesisRole-${AWS::StackName}
      Path: /

  KinesisWritePolicy:
    DependsOn:
      - DmsKinesisRole
      - KinesisDataStreamCdc
    Type: AWS::IAM::Policy
    Properties:
      PolicyName: !Sub KinesisDataPolicy-${AWS::StackName}
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          -
            Effect: "Allow"
            Action:
              - kinesis:DescribeStream
              - kinesis:PutRecord
              - kinesis:PutRecords
            Resource: !GetAtt KinesisDataStreamCdc.Arn
      Roles:
        - !Ref DmsKinesisRole

  DmsSourceEndpoint:
    DependsOn:
      - RDS
    Type: AWS::DMS::Endpoint
    Properties:
      EndpointType: source
      EndpointIdentifier: !Sub "source-${AWS::StackName}"
      EngineName: mysql
      ServerName: !GetAtt RDS.Endpoint.Address
      Port: !GetAtt RDS.Endpoint.Port
      Username: !Ref DatabaseUserName
      Password: !Ref DatabaseUserPassword

  DmsTargetEndpoint:
    DependsOn:
      - DmsKinesisRole
      - KinesisDataStreamCdc
      - KinesisWritePolicy
    Type: AWS::DMS::Endpoint
    Properties:
      EndpointType: target
      EndpointIdentifier: !Sub "target-${AWS::StackName}"
      EngineName: kinesis
      KinesisSettings:
        StreamArn: !GetAtt KinesisDataStreamCdc.Arn
        ServiceAccessRoleArn: !GetAtt DmsKinesisRole.Arn
        MessageFormat: json

  DmsReplicationInstance:
    DependsOn:
      - RDS
      - SecurityGroup
      - DmsReplicationSubnetGroup
    Type: AWS::DMS::ReplicationInstance
    Properties:
      ReplicationInstanceIdentifier: !Sub "repl-${AWS::StackName}"
      ReplicationInstanceClass: dms.t3.medium
      EngineVersion: 3.4.7
      AllocatedStorage: 100
      PubliclyAccessible: False
      ReplicationSubnetGroupIdentifier: !Ref DmsReplicationSubnetGroup
      VpcSecurityGroupIds:
        - !GetAtt SecurityGroup.GroupId

  DmsReplicationSubnetGroup:
    Type: AWS::DMS::ReplicationSubnetGroup
    Properties:
      ReplicationSubnetGroupDescription: !Sub "cdc-sample-${AWS::StackName}"
      SubnetIds:
        - !Ref PrivateSubnetA
        - !Ref PrivateSubnetB

  DmsReplicationTask:
    DependsOn:
      - DmsSourceEndpoint
      - DmsTargetEndpoint
      - DmsReplicationInstance
    Type: AWS::DMS::ReplicationTask
    Properties:
      ReplicationTaskIdentifier: !Sub "rds-to-kinesis-${AWS::StackName}"
      SourceEndpointArn: !Ref DmsSourceEndpoint
      TargetEndpointArn: !Ref DmsTargetEndpoint
      ReplicationInstanceArn: !Ref DmsReplicationInstance
      MigrationType: cdc
      TableMappings: "{\"rules\":[{\"rule-type\":\"selection\",\"rule-id\":\"277261178\",\"rule-name\":\"277261178\",\"object-locator\":{\"schema-name\":\"dev\",\"table-name\":\"ticket\"},\"rule-action\":\"include\",\"filters\":[]},{\"rule-type\":\"selection\",\"rule-id\":\"277260823\",\"rule-name\":\"277260823\",\"object-locator\":{\"schema-name\":\"dev\",\"table-name\":\"customer\"},\"rule-action\":\"include\",\"filters\":[]},{\"rule-type\":\"selection\",\"rule-id\":\"277260359\",\"rule-name\":\"277260359\",\"object-locator\":{\"schema-name\":\"dev\",\"table-name\":\"sport_event\"},\"rule-action\":\"include\",\"filters\":[]},{\"rule-type\":\"selection\",\"rule-id\":\"276358336\",\"rule-name\":\"276358336\",\"object-locator\":{\"schema-name\":\"dev\",\"table-name\":\"ticket_activity\"},\"rule-action\":\"include\",\"filters\":[]}]}"

  GlueDatabase:
    Type: AWS::Glue::Database
    Properties:
      CatalogId: !Ref AWS::AccountId
      DatabaseInput:
        Name: !Sub "database_${AWS::AccountId}_hudi_cdc_demo"

  GlueTableKinesis:
    DependsOn:
      - GlueDatabase
      - KinesisDataStreamCdc
    Type: AWS::Glue::Table
    Properties:
      CatalogId: !Ref AWS::AccountId
      DatabaseName: !Ref GlueDatabase
      TableInput:
        Name: !Sub "kinesis_cdc_table"
        TableType: EXTERNAL_TABLE
        Parameters: {
          "classification": "json"
        }
        StorageDescriptor:
          Location: !Ref KinesisDataStreamCdc
          InputFormat: org.apache.hadoop.mapred.TextInputFormat
          OutputFormat: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat          
          SerdeInfo:
            SerializationLibrary: org.openx.data.jsonserde.JsonSerDe
            Parameters: 
                paths: ""
          Parameters:
            typeOfData: kinesis
            streamName: !Ref KinesisDataStreamCdc
            endpointUrl: !Sub 'https://kinesis.${AWS::Region}.amazonaws.com'

  RDSSecret:
    DependsOn:
      - RDS
    Type: AWS::SecretsManager::Secret
    Properties:
      Name: !Sub "${AWS::StackName}-my-secret"
      SecretString:
        !Join
        - ""
        - - "{\"db_user\":\""
          - !Ref DatabaseUserName
          - "\",\"db_password\":\""
          - !Ref DatabaseUserPassword
          - "\",\"db_host\":\""
          - !GetAtt RDS.Endpoint.Address
          - "\",\"db_port\":\""
          - !GetAtt RDS.Endpoint.Port
          - "\",\"db_engine\":\""
          - "mysql"
          - "\",\"db_database\":\""
          - "dev"
          - "\"}"