From caaefc0c0b663212a36807b1ebb1d18b76a0ae6c Mon Sep 17 00:00:00 2001
From: Andrey <andrey.sorokin@akvelon.com>
Date: Thu, 21 Jul 2022 06:11:35 +0600
Subject: [PATCH] [CdapIO] HasOffset interface was implemented (#22193)

* [BEAM-22148] HasOffset interface was implemented

* [BEAM-22148] Formatting was fixed
---
 sdks/java/io/sparkreceiver/build.gradle       |  2 ++
 .../beam/sdk/io/sparkreceiver/HasOffset.java  | 31 +++++++++++++++++++
 settings.gradle.kts                           |  1 +
 3 files changed, 34 insertions(+)
 create mode 100644 sdks/java/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/HasOffset.java

diff --git a/sdks/java/io/sparkreceiver/build.gradle b/sdks/java/io/sparkreceiver/build.gradle
index 09f98db124631..f226435631e4c 100644
--- a/sdks/java/io/sparkreceiver/build.gradle
+++ b/sdks/java/io/sparkreceiver/build.gradle
@@ -41,4 +41,6 @@ dependencies {
     compileOnly "org.scala-lang:scala-library:2.11.12"
     testImplementation project(path: ":sdks:java:io:cdap", configuration: "testRuntimeMigration")
     testImplementation library.java.junit
+    testImplementation project(path: ":runners:direct-java", configuration: "shadow")
+    testImplementation project(path: ":examples:java", configuration: "testRuntimeMigration")
 }
diff --git a/sdks/java/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/HasOffset.java b/sdks/java/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/HasOffset.java
new file mode 100644
index 0000000000000..b3c9e11defdbc
--- /dev/null
+++ b/sdks/java/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/HasOffset.java
@@ -0,0 +1,31 @@
+/*
+ * 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.sdk.io.sparkreceiver;
+
+/**
+ * Interface for any Spark {@link org.apache.spark.streaming.receiver.Receiver} that supports
+ * reading from and to some offset.
+ */
+public interface HasOffset {
+
+  /** @param offset inclusive start offset from which the reading should be started. */
+  void setStartOffset(Long offset);
+
+  /** @return exclusive end offset to which the reading from current page will occur. */
+  Long getEndOffset();
+}
diff --git a/settings.gradle.kts b/settings.gradle.kts
index 05ecc694733f5..5a35c29f2e1fa 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -194,6 +194,7 @@ include(":sdks:java:io:pulsar")
 include(":sdks:java:io:rabbitmq")
 include(":sdks:java:io:redis")
 include(":sdks:java:io:solr")
+include(":sdks:java:io:sparkreceiver")
 include(":sdks:java:io:snowflake")
 include(":sdks:java:io:snowflake:expansion-service")
 include(":sdks:java:io:splunk")