diff --git a/async-shuffle-upload/async-shuffle-upload-api/pom.xml b/async-shuffle-upload/async-shuffle-upload-api/pom.xml new file mode 100644 index 0000000000000..399e527239595 --- /dev/null +++ b/async-shuffle-upload/async-shuffle-upload-api/pom.xml @@ -0,0 +1,69 @@ + + + + 4.0.0 + + org.apache.spark + spark-parent_2.11 + 3.0.0-SNAPSHOT + ../../pom.xml + + + spark-async-shuffle-upload-api + + spark-async-shuffle-upload-api + + jar + Spark Project Async Shuffle Upload API + http://spark.apache.org/ + + + org.immutables + value + + + com.fasterxml.jackson.datatype + jackson-datatype-jdk8 + + + com.palantir.safe-logging + safe-logging + + + com.palantir.safe-logging + preconditions + + + org.apache.spark + spark-async-shuffle-upload-immutables + ${project.version} + provided + + + org.assertj + assertj-core + test + + + junit + junit + test + + + + diff --git a/async-shuffle-upload/async-shuffle-upload-api/src/main/java/com/palantir/spark/shuffle/async/api/SparkShuffleApiConstants.java b/async-shuffle-upload/async-shuffle-upload-api/src/main/java/com/palantir/spark/shuffle/async/api/SparkShuffleApiConstants.java new file mode 100644 index 0000000000000..36a7ad089712b --- /dev/null +++ b/async-shuffle-upload/async-shuffle-upload-api/src/main/java/com/palantir/spark/shuffle/async/api/SparkShuffleApiConstants.java @@ -0,0 +1,32 @@ +/* + * 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 com.palantir.spark.shuffle.async.api; + +public final class SparkShuffleApiConstants { + + // Identifiers used by the spark shuffle plugin + // TODO make these consistent in naming - but keep in mind that changing these would result in runtime errors. + public static final String SHUFFLE_PLUGIN_APP_NAME_CONF = "spark.plugin.shuffle.async.appName"; + public static final String SHUFFLE_BASE_URI_CONF = "spark.shuffle.hadoop.async.base-uri"; + public static final String SHUFFLE_S3A_CREDS_FILE_CONF = "spark.plugin.shuffle.async.s3a.credsFile"; + public static final String SHUFFLE_S3A_ENDPOINT_CONF = "spark.shuffle.hadoop.async.s3a.endpoint"; + public static final String METRICS_FACTORY_CLASS_CONF = "spark.plugin.shuffle.async.metricsFactoryClass"; + + private SparkShuffleApiConstants() {} + +} diff --git a/async-shuffle-upload/async-shuffle-upload-api/src/main/java/com/palantir/spark/shuffle/async/api/SparkShuffleAwsCredentials.java b/async-shuffle-upload/async-shuffle-upload-api/src/main/java/com/palantir/spark/shuffle/async/api/SparkShuffleAwsCredentials.java new file mode 100644 index 0000000000000..a995c3de105fb --- /dev/null +++ b/async-shuffle-upload/async-shuffle-upload-api/src/main/java/com/palantir/spark/shuffle/async/api/SparkShuffleAwsCredentials.java @@ -0,0 +1,70 @@ +/* + * 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 com.palantir.spark.shuffle.async.api; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.palantir.logsafe.UnsafeArg; +import com.palantir.logsafe.exceptions.SafeIllegalArgumentException; +import com.palantir.spark.shuffle.async.immutables.ImmutablesStyle; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import org.immutables.value.Value; + +@ImmutablesStyle +@Value.Immutable +@JsonSerialize(as = ImmutableSparkShuffleAwsCredentials.class) +@JsonDeserialize(as = ImmutableSparkShuffleAwsCredentials.class) +@JsonIgnoreProperties(ignoreUnknown = true) +public abstract class SparkShuffleAwsCredentials { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + public abstract String accessKeyId(); + + public abstract String secretAccessKey(); + + public abstract String sessionToken(); + + public final byte[] toBytes() { + try { + return MAPPER.writeValueAsString(this).getBytes(StandardCharsets.UTF_8); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + + public static SparkShuffleAwsCredentials fromBytes(byte[] bytes) { + try { + return MAPPER.readValue(new String(bytes, StandardCharsets.UTF_8), SparkShuffleAwsCredentials.class); + } catch (IOException e) { + throw new SafeIllegalArgumentException( + "Could not deserialize bytes as AWS credentials.", + UnsafeArg.of("cause", e)); + } + } + + public static Builder builder() { + return new Builder(); + } + + public static final class Builder extends ImmutableSparkShuffleAwsCredentials.Builder {} +} diff --git a/async-shuffle-upload/async-shuffle-upload-api/src/test/java/com/palantir/spark/shuffle/async/api/SparkShuffleAwsCredentialsSuite.java b/async-shuffle-upload/async-shuffle-upload-api/src/test/java/com/palantir/spark/shuffle/async/api/SparkShuffleAwsCredentialsSuite.java new file mode 100644 index 0000000000000..8c80146ea7f41 --- /dev/null +++ b/async-shuffle-upload/async-shuffle-upload-api/src/test/java/com/palantir/spark/shuffle/async/api/SparkShuffleAwsCredentialsSuite.java @@ -0,0 +1,51 @@ +/* + * 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 com.palantir.spark.shuffle.async.api; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; + +import java.nio.charset.StandardCharsets; +import org.junit.Test; + +public final class SparkShuffleAwsCredentialsSuite { + + @Test + public void testSerialize() { + SparkShuffleAwsCredentials creds = SparkShuffleAwsCredentials.builder() + .accessKeyId("access-key") + .secretAccessKey("secret-key") + .sessionToken("session-token") + .build(); + byte[] bytes = creds.toBytes(); + assertThat(new String(bytes, StandardCharsets.UTF_8)) + .isEqualTo("{\"accessKeyId\":\"access-key\"," + + "\"secretAccessKey\":\"secret-key\"," + + "\"sessionToken\":\"session-token\"}"); + } + + @Test + public void testDeserialize() { + String serializedString = "{\"accessKeyId\":\"access-key\"," + + "\"secretAccessKey\":\"secret-key\"," + + "\"sessionToken\":\"session-token\"}"; + + SparkShuffleAwsCredentials creds = + SparkShuffleAwsCredentials.fromBytes(serializedString.getBytes(StandardCharsets.UTF_8)); + } +} diff --git a/async-shuffle-upload/immutables/pom.xml b/async-shuffle-upload/immutables/pom.xml new file mode 100644 index 0000000000000..67129a4a9908a --- /dev/null +++ b/async-shuffle-upload/immutables/pom.xml @@ -0,0 +1,41 @@ + + + + + 4.0.0 + + org.apache.spark + spark-parent_2.11 + 3.0.0-SNAPSHOT + ../../pom.xml + + + spark-async-shuffle-upload-immutables + + spark-async-shuffle-upload-immutables + + jar + Spark Project Async Shuffle Upload Immutables + http://spark.apache.org/ + + + org.immutables + value + + + diff --git a/async-shuffle-upload/immutables/src/main/java/com/palantir/spark/shuffle/async/immutables/ImmutablesStyle.java b/async-shuffle-upload/immutables/src/main/java/com/palantir/spark/shuffle/async/immutables/ImmutablesStyle.java new file mode 100644 index 0000000000000..6cdd7b30c1fea --- /dev/null +++ b/async-shuffle-upload/immutables/src/main/java/com/palantir/spark/shuffle/async/immutables/ImmutablesStyle.java @@ -0,0 +1,33 @@ +/* + * 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 com.palantir.spark.shuffle.async.immutables; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import org.immutables.value.Value; + +@Target({ElementType.PACKAGE, ElementType.TYPE}) +@Retention(RetentionPolicy.SOURCE) +@Value.Style( + visibility = Value.Style.ImplementationVisibility.PACKAGE, + overshadowImplementation = true, + jdkOnly = true, + get = {"get*", "is*"}) +public @interface ImmutablesStyle {} diff --git a/core/pom.xml b/core/pom.xml index 3c2bae47fa5cf..c533853d427d9 100644 --- a/core/pom.xml +++ b/core/pom.xml @@ -388,14 +388,13 @@ test - com.palantir.safe-logging - safe-logging - 1.5.1 + org.apache.commons + commons-crypto - org.apache.commons - commons-crypto + com.palantir.safe-logging + safe-logging diff --git a/pom.xml b/pom.xml index 56050c89ea471..3675daea8322a 100644 --- a/pom.xml +++ b/pom.xml @@ -114,6 +114,10 @@ sql/hive streaming tools + + + async-shuffle-upload/immutables + async-shuffle-upload/async-shuffle-upload-api @@ -216,6 +220,9 @@ --> 0.12.0 + + 1.13.0 + ${java.home} @@ -2305,6 +2312,30 @@ kafka-clients ${kafka.version} + + + com.palantir.safe-logging + safe-logging + ${safe-logging.version} + + + + + org.immutables + value + 2.8.2 + provided + + + com.palantir.safe-logging + preconditions + ${safe-logging.version} + + + org.assertj + assertj-core + 3.15.0 +