YuriGusev commented on code in PR #1: URL: https://github.com/apache/flink-connector-dynamodb/pull/1#discussion_r1008763069
########## flink-connector-dynamodb/src/main/java/org/apache/flink/streaming/connectors/dynamodb/util/PrimaryKeyBuilder.java: ########## @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.streaming.connectors.dynamodb.util; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.dynamodb.sink.InvalidRequestException; +import org.apache.flink.util.CollectionUtil; + +import software.amazon.awssdk.services.dynamodb.model.AttributeValue; +import software.amazon.awssdk.services.dynamodb.model.WriteRequest; + +import java.util.List; +import java.util.Map; +import java.util.UUID; + +/** Helper to construct primary(composite) key for a DynamoDB request. */ +@Internal +public class PrimaryKeyBuilder { + + public static String build(List<String> pKeys, WriteRequest request) { + + if (CollectionUtil.isNullOrEmpty(pKeys)) { + // fake key, because no dynamodb partition key or sort key provided. Using UUID should be safe + // here, as we have at most 25 elements per batch + return UUID.randomUUID().toString(); + } else { + Map<String, AttributeValue> requestItems = getRequestItems(request); + + StringBuilder builder = new StringBuilder(); + for (String keyName : pKeys) { + AttributeValue value = requestItems.get(keyName); + + if (value == null) { + throw new InvalidRequestException( + "Request " + request.toString() + " does not contain pKey " + keyName); + } + + builder.append(getKeyValue(value)); + } + + return builder.toString(); + } + } + + /** + * Returns string value of a partition key attribute. Each primary key attribute must be defined + * as type String, Number, or binary as per DynamoDB specification. + */ + private static String getKeyValue(AttributeValue value) { + StringBuilder builder = new StringBuilder(); + + if (value.n() != null) { + builder.append(value.n()); + } + + if (value.s() != null) { + builder.append(value.s()); + } + + if (value.b() != null) { + builder.append(value.b().asUtf8String()); + } + + return builder.toString(); Review Comment: I added validation as per DynamoDB specification partition key, and sort key values can not be blank. However, we would only be able to validate that if the deduplication configuration is provided. When not provided, the request will be failed by DynamoDB. DynamoDB would fail the request if we do not validate for the blank values at all, but I agree it is good to fail fast. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
