kafka-connect-s3 SSE without customer key broken

[85danf]

Versions:

• 10.5.6
• 10.4.8
• 10.3.10
• 10.2.14
• 10.1.15
• 10.0.26

Hey, we've been happily using kafka-connect-s3 for a long time and recently after a version upgrade one of our usecases started to break:
It's a bucket not owned by my team which has AES256 SSE configured with an aws-provided key (not customer provided).
up until the version upgrade it was working perfectly and since the upgrade we started getting permission errors like

Caused by: com.amazonaws.services.s3.model.AmazonS3Exception: User: arn:aws:sts::***:assumed-role/***/aws-sdk-java-*** is not authorized 
to perform: kms:GenerateDataKey on this resource because the resource does not exist in this Region, no resource-based policies allow access, 
or a resource-based policy explicitly denies access (Service: Amazon S3; Status Code: 403; Error Code: AccessDenied; Request ID: ***; 
S3 Extended Request ID: ***; Proxy: null), S3 Extended Request ID: ***

I've pinpointed the issue to this commit / PR.

Looks like its not taking this usecase under consideration and always expects a customer key to be available, which is something that is taken into account when initializing the customer key in S3OutputStream::79 :

this.sseCustomerKey = (SSEAlgorithm.AES256.toString().equalsIgnoreCase(ssea)
        && StringUtils.isNotBlank(sseCustomerKeyConfig))
      ? new SSECustomerKey(sseCustomerKeyConfig) : null;

I think the x-amz-server-side-encryption header still has to be included when ssea == AES256 and sseCustomerKey == null

Related:
#389
#173
#684
10b020a