private static void createDeliveryStream ( ) throws Exception { boolean deliveryStreamExists = false ; LOG . info ( "Checking if " + deliveryStreamName + " already exits" ) ; List < String > deliveryStreamNames = listDeliveryStreams ( ) ; if ( deliveryStreamNames != null && deliveryStreamNames . contains ( deliveryStreamName ) ) { deliveryStreamExists = true ; LOG . info ( "DeliveryStream " + deliveryStreamName + " already exists. Not creating the new delivery stream" ) ; } else { LOG . info ( "DeliveryStream " + deliveryStreamName + " does not exist" ) ; } if ( ! deliveryStreamExists ) { // Create deliveryStream CreateDeliveryStreamRequest createDeliveryStreamRequest = new CreateDeliveryStreamRequest ( ) ; createDeliveryStreamRequest . setDeliveryStreamName ( deliveryStreamName ) ; S3DestinationConfiguration s3DestinationConfiguration = new S3DestinationConfiguration ( ) ; s3DestinationConfiguration . setBucketARN ( s3BucketARN ) ; s3DestinationConfiguration . setPrefix ( s3ObjectPrefix ) ; // Could also specify GZIP or ZIP s3DestinationConfiguration . setCompressionFormat ( CompressionFormat . UNCOMPRESSED ) ; // Encryption configuration is optional EncryptionConfiguration encryptionConfiguration = new EncryptionConfiguration ( ) ; if ( ! StringUtils . isNullOrEmpty ( s3DestinationAWSKMSKeyId ) ) { encryptionConfiguration . setKMSEncryptionConfig ( new KMSEncryptionConfig ( ) . withAWSKMSKeyARN ( s3DestinationAWSKMSKeyId ) ) ; } else { encryptionConfiguration . setNoEncryptionConfig ( NoEncryptionConfig . NoEncryption ) ; } s3DestinationConfiguration . setEncryptionConfiguration ( encryptionConfiguration ) ; BufferingHints bufferingHints = null ; if ( s3DestinationSizeInMBs != null || s3DestinationIntervalInSeconds != null ) { bufferingHints = new BufferingHints ( ) ; bufferingHints . setSizeInMBs ( s3DestinationSizeInMBs ) ; bufferingHints . setIntervalInSeconds ( s3DestinationIntervalInSeconds ) ; } s3DestinationConfiguration . setBufferingHints ( bufferingHints ) ; // Create and set IAM role so that firehose service has access to the S3Buckets to put data // and KMS keys (if provided) to encrypt data. Please check the trustPolicyDocument.json and // permissionsPolicyDocument.json files for the trust and permissions policies set for the role. String iamRoleArn = createIamRole ( s3ObjectPrefix ) ; s3DestinationConfiguration . setRoleARN ( iamRoleArn ) ; createDeliveryStreamRequest . setS3DestinationConfiguration ( s3DestinationConfiguration ) ; firehoseClient . createDeliveryStream ( createDeliveryStreamRequest ) ; // The Delivery Stream is now being created. LOG . info ( "Creating DeliveryStream : " + deliveryStreamName ) ; waitForDeliveryStreamToBecomeAvailable ( deliveryStreamName ) ; } }
Method to create delivery stream for S3 destination configuration .