This is an automated email from the ASF dual-hosted git repository. damondouglas pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push: new 104c10b3ee5 [RRIO] Create Caller and SetupTeardown interfaces (#28905) 104c10b3ee5 is described below commit 104c10b3ee536a9a3ea52b4dbf62d86b669da5d9 Author: Damon <damondoug...@users.noreply.github.com> AuthorDate: Wed Oct 11 12:39:38 2023 -0700 [RRIO] Create Caller and SetupTeardown interfaces (#28905) * Create test Caller and SetupTeardown interfaces * Update Javadoc * Defer Call transform to future PR * Rename package to requestresponseio * Add username to TODO --- sdks/java/io/rrio/build.gradle | 7 +- .../apache/beam/io/requestresponseio/Caller.java | 27 +++++ .../beam/io/requestresponseio/SetupTeardown.java | 34 ++++++ .../UserCodeExecutionException.java | 38 ++++++ .../requestresponseio/UserCodeQuotaException.java | 42 +++++++ .../UserCodeTimeoutException.java | 39 ++++++ .../beam/io/requestresponseio/package-info.java | 20 ++++ .../java/org/apache/beam/io/rrio/CallerTest.java | 126 ++++++++++++++++++++ .../org/apache/beam/io/rrio/SetupTeardownTest.java | 132 +++++++++++++++++++++ 9 files changed, 462 insertions(+), 3 deletions(-) diff --git a/sdks/java/io/rrio/build.gradle b/sdks/java/io/rrio/build.gradle index d65df370e0c..32fbd9d22e3 100644 --- a/sdks/java/io/rrio/build.gradle +++ b/sdks/java/io/rrio/build.gradle @@ -25,9 +25,10 @@ description = "Apache Beam :: SDKS :: Java :: IO :: RequestResponseIO (RRIO)" ext.summary = "Support to read from and write to Web APIs" dependencies { - implementation project(path: ":sdks:java:core", configuration: "shadow") - implementation library.java.joda_time - implementation library.java.vendored_guava_32_1_2_jre + // TODO(damondouglas): revert to implementation after project is more fully developed + permitUnusedDeclared project(path: ":sdks:java:core", configuration: "shadow") + permitUnusedDeclared library.java.joda_time + permitUnusedDeclared library.java.vendored_guava_32_1_2_jre testImplementation project(path: ":sdks:java:core", configuration: "shadowTest") testImplementation library.java.junit diff --git a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/Caller.java b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/Caller.java new file mode 100644 index 00000000000..32b514c43a1 --- /dev/null +++ b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/Caller.java @@ -0,0 +1,27 @@ +/* + * 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.beam.io.requestresponseio; + +import java.io.Serializable; + +/** {@link Caller} interfaces user custom code intended for API calls. */ +public interface Caller<RequestT, ResponseT> extends Serializable { + + /** Calls a Web API with the {@link RequestT} and returns a {@link ResponseT}. */ + ResponseT call(RequestT request) throws UserCodeExecutionException; +} diff --git a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/SetupTeardown.java b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/SetupTeardown.java new file mode 100644 index 00000000000..2bdc8113d98 --- /dev/null +++ b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/SetupTeardown.java @@ -0,0 +1,34 @@ +/* + * 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.beam.io.requestresponseio; + +import java.io.Serializable; + +/** + * Provided by user and called within {@link org.apache.beam.sdk.transforms.DoFn.Setup} and @{link + * org.apache.beam.sdk.transforms.DoFn.Teardown} lifecycle methods of {@link Call}'s {@link + * org.apache.beam.sdk.transforms.DoFn}. + */ +public interface SetupTeardown extends Serializable { + + /** Called during the {@link org.apache.beam.sdk.transforms.DoFn}'s setup lifecycle method. */ + void setup() throws UserCodeExecutionException; + + /** Called during the {@link org.apache.beam.sdk.transforms.DoFn}'s teardown lifecycle method. */ + void teardown() throws UserCodeExecutionException; +} diff --git a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/UserCodeExecutionException.java b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/UserCodeExecutionException.java new file mode 100644 index 00000000000..3a4c002f52e --- /dev/null +++ b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/UserCodeExecutionException.java @@ -0,0 +1,38 @@ +/* + * 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.beam.io.requestresponseio; + +/** Base {@link Exception} for signaling errors in user custom code. */ +public class UserCodeExecutionException extends Exception { + public UserCodeExecutionException(String message) { + super(message); + } + + public UserCodeExecutionException(String message, Throwable cause) { + super(message, cause); + } + + public UserCodeExecutionException(Throwable cause) { + super(cause); + } + + public UserCodeExecutionException( + String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) { + super(message, cause, enableSuppression, writableStackTrace); + } +} diff --git a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/UserCodeQuotaException.java b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/UserCodeQuotaException.java new file mode 100644 index 00000000000..f16f078927f --- /dev/null +++ b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/UserCodeQuotaException.java @@ -0,0 +1,42 @@ +/* + * 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.beam.io.requestresponseio; + +/** + * Extends {@link UserCodeQuotaException} to allow the user custom code to specifically signal a + * Quota or API overuse related error. + */ +public class UserCodeQuotaException extends UserCodeExecutionException { + + public UserCodeQuotaException(String message) { + super(message); + } + + public UserCodeQuotaException(String message, Throwable cause) { + super(message, cause); + } + + public UserCodeQuotaException(Throwable cause) { + super(cause); + } + + public UserCodeQuotaException( + String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) { + super(message, cause, enableSuppression, writableStackTrace); + } +} diff --git a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/UserCodeTimeoutException.java b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/UserCodeTimeoutException.java new file mode 100644 index 00000000000..22b06744985 --- /dev/null +++ b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/UserCodeTimeoutException.java @@ -0,0 +1,39 @@ +/* + * 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.beam.io.requestresponseio; + +/** An extension of {@link UserCodeQuotaException} to specifically signal a user code timeout. */ +public class UserCodeTimeoutException extends UserCodeExecutionException { + + public UserCodeTimeoutException(String message) { + super(message); + } + + public UserCodeTimeoutException(String message, Throwable cause) { + super(message, cause); + } + + public UserCodeTimeoutException(Throwable cause) { + super(cause); + } + + public UserCodeTimeoutException( + String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) { + super(message, cause, enableSuppression, writableStackTrace); + } +} diff --git a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/package-info.java b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/package-info.java new file mode 100644 index 00000000000..cd9c11c13f8 --- /dev/null +++ b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponseio/package-info.java @@ -0,0 +1,20 @@ +/* + * 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 provides Beam I/O transform support for safely reading from and writing to Web APIs. */ +package org.apache.beam.io.requestresponseio; diff --git a/sdks/java/io/rrio/src/test/java/org/apache/beam/io/rrio/CallerTest.java b/sdks/java/io/rrio/src/test/java/org/apache/beam/io/rrio/CallerTest.java new file mode 100644 index 00000000000..5258573f428 --- /dev/null +++ b/sdks/java/io/rrio/src/test/java/org/apache/beam/io/rrio/CallerTest.java @@ -0,0 +1,126 @@ +/* + * 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.beam.io.rrio; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + +import org.apache.beam.io.requestresponseio.Caller; +import org.apache.beam.io.requestresponseio.UserCodeExecutionException; +import org.apache.beam.io.requestresponseio.UserCodeQuotaException; +import org.apache.beam.io.requestresponseio.UserCodeTimeoutException; +import org.apache.beam.sdk.Pipeline.PipelineExecutionException; +import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.util.SerializableUtils; +import org.joda.time.Instant; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests for {@link Caller}. */ +@RunWith(JUnit4.class) +public class CallerTest { + + @Rule public TestPipeline pipeline = TestPipeline.create(); + + @Test + public void canSerializeImplementingClasses() { + SerializableUtils.serializeToByteArray(new CallerImpl()); + } + + @Test + public void canSerializeWhenUsedInDoFn() { + pipeline + .apply(Create.of(Instant.now())) + .apply(ParDo.of(new CallerUsingDoFn<>(new CallerImpl()))) + .setCoder(StringUtf8Coder.of()); + + pipeline.run(); + } + + @Test + public void canSignalQuotaException() { + pipeline + .apply(Create.of(1)) + .apply(ParDo.of(new CallerUsingDoFn<>(new CallerThrowsQuotaException()))) + .setCoder(VarIntCoder.of()); + + PipelineExecutionException executionException = + assertThrows(PipelineExecutionException.class, pipeline::run); + assertEquals(UserCodeQuotaException.class, executionException.getCause().getClass()); + } + + @Test + public void canSignalTimeoutException() { + pipeline + .apply(Create.of(1)) + .apply(ParDo.of(new CallerUsingDoFn<>(new CallerThrowsTimeoutException()))) + .setCoder(VarIntCoder.of()); + + PipelineExecutionException executionException = + assertThrows(PipelineExecutionException.class, pipeline::run); + assertEquals(UserCodeTimeoutException.class, executionException.getCause().getClass()); + } + + private static class CallerUsingDoFn<RequestT, ResponseT> extends DoFn<RequestT, ResponseT> { + private final Caller<RequestT, ResponseT> caller; + + private CallerUsingDoFn(Caller<RequestT, ResponseT> caller) { + this.caller = caller; + } + + @ProcessElement + public void process(@Element RequestT request, OutputReceiver<ResponseT> receiver) + throws UserCodeExecutionException { + RequestT safeRequest = checkStateNotNull(request); + ResponseT response = caller.call(safeRequest); + receiver.output(response); + } + } + + private static class CallerImpl implements Caller<Instant, String> { + + @Override + public String call(Instant request) throws UserCodeExecutionException { + return request.toString(); + } + } + + private static class CallerThrowsQuotaException implements Caller<Integer, Integer> { + + @Override + public Integer call(Integer request) throws UserCodeExecutionException { + throw new UserCodeQuotaException("quota"); + } + } + + private static class CallerThrowsTimeoutException implements Caller<Integer, Integer> { + + @Override + public Integer call(Integer request) throws UserCodeExecutionException { + throw new UserCodeTimeoutException("timeout"); + } + } +} diff --git a/sdks/java/io/rrio/src/test/java/org/apache/beam/io/rrio/SetupTeardownTest.java b/sdks/java/io/rrio/src/test/java/org/apache/beam/io/rrio/SetupTeardownTest.java new file mode 100644 index 00000000000..a8c5c45ede5 --- /dev/null +++ b/sdks/java/io/rrio/src/test/java/org/apache/beam/io/rrio/SetupTeardownTest.java @@ -0,0 +1,132 @@ +/* + * 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.beam.io.rrio; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + +import org.apache.beam.io.requestresponseio.SetupTeardown; +import org.apache.beam.io.requestresponseio.UserCodeExecutionException; +import org.apache.beam.io.requestresponseio.UserCodeQuotaException; +import org.apache.beam.io.requestresponseio.UserCodeTimeoutException; +import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.util.SerializableUtils; +import org.apache.beam.sdk.util.UserCodeException; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.UncheckedExecutionException; +import org.junit.Rule; +import org.junit.Test; + +public class SetupTeardownTest { + @Rule public TestPipeline pipeline = TestPipeline.create(); + + @Test + public void canSerializeImplementingClasses() { + SerializableUtils.serializeToByteArray(new SetupTeardownImpl()); + } + + @Test + public void canSerializeWhenUsedInDoFn() { + pipeline + .apply(Create.of(1)) + .apply(ParDo.of(new SetupTeardownUsingDoFn(new SetupTeardownImpl()))) + .setCoder(VarIntCoder.of()); + + pipeline.run(); + } + + @Test + public void canSignalQuotaException() { + pipeline + .apply(Create.of(1)) + .apply(ParDo.of(new SetupTeardownUsingDoFn(new ThrowsQuotaException()))) + .setCoder(VarIntCoder.of()); + + UncheckedExecutionException exception = + assertThrows(UncheckedExecutionException.class, pipeline::run); + UserCodeException userCodeException = (UserCodeException) exception.getCause(); + assertEquals(UserCodeQuotaException.class, userCodeException.getCause().getClass()); + } + + @Test + public void canSignalTimeoutException() { + pipeline + .apply(Create.of(1)) + .apply(ParDo.of(new SetupTeardownUsingDoFn(new ThrowsTimeoutException()))) + .setCoder(VarIntCoder.of()); + + UncheckedExecutionException exception = + assertThrows(UncheckedExecutionException.class, pipeline::run); + UserCodeException userCodeException = (UserCodeException) exception.getCause(); + assertEquals(UserCodeTimeoutException.class, userCodeException.getCause().getClass()); + } + + private static class SetupTeardownUsingDoFn extends DoFn<Integer, Integer> { + private final SetupTeardown setupTeardown; + + private SetupTeardownUsingDoFn(SetupTeardown setupTeardown) { + this.setupTeardown = setupTeardown; + } + + @Setup + public void setup() throws UserCodeExecutionException { + setupTeardown.setup(); + } + + @Teardown + public void teardown() throws UserCodeExecutionException { + setupTeardown.teardown(); + } + + @ProcessElement + public void process() {} + } + + private static class SetupTeardownImpl implements SetupTeardown { + @Override + public void setup() throws UserCodeExecutionException {} + + @Override + public void teardown() throws UserCodeExecutionException {} + } + + private static class ThrowsQuotaException implements SetupTeardown { + + @Override + public void setup() throws UserCodeExecutionException { + throw new UserCodeQuotaException("quota"); + } + + @Override + public void teardown() throws UserCodeExecutionException {} + } + + private static class ThrowsTimeoutException implements SetupTeardown { + + @Override + public void setup() throws UserCodeExecutionException { + throw new UserCodeTimeoutException("timeout"); + } + + @Override + public void teardown() throws UserCodeExecutionException {} + } +}