Moved Record interface as part of functions api (apache#2184)

* Moved Record interface as part of functions api

* Added org.apache.pulsar.functions.api.* to PowerMock ignore

pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java

@@ -56,7 +56,7 @@
 
 @Slf4j
 @PrepareForTest({CmdFunctions.class})
-@PowerMockIgnore({ "javax.management.*", "javax.ws.*", "org.apache.logging.log4j.*", "org.apache.pulsar.io.core.*" })
+@PowerMockIgnore({ "javax.management.*", "javax.ws.*", "org.apache.logging.log4j.*", "org.apache.pulsar.io.core.*", "org.apache.pulsar.functions.api.*" })
 public class TestCmdSinks {
 
     @ObjectFactory
@@ -155,7 +155,7 @@ public void testCliCorrect() throws Exception {
                 sinkConfig
         );
     }
-    
+
     @Test
     public void testMissingTenant() throws Exception {
         SinkConfig sinkConfig = getSinkConfig();
@@ -816,8 +816,8 @@ public void testCmdSinkConfigFile(SinkConfig testSinkConfig, SinkConfig expected
         verify(updateSink).validateSinkConfigs(eq(expectedSinkConfig));
         verify(localSinkRunner).validateSinkConfigs(eq(expectedSinkConfig));
     }
-    
-    
+
+
     @Test
     public void testCliOverwriteConfigFile() throws Exception {
 
@@ -837,7 +837,7 @@ public void testCliOverwriteConfigFile() throws Exception {
         testSinkConfig.setArchive(JAR_FILE_PATH + "-prime");
         testSinkConfig.setResources(new Resources(CPU + 1, RAM + 1, DISK + 1));
         testSinkConfig.setConfigs(createSink.parseConfigs("{\"created_at-prime\":\"Mon Jul 02 00:33:15 +0000 2018\"}"));
-        
+
 
         SinkConfig expectedSinkConfig = getSinkConfig();
 
@@ -846,7 +846,7 @@ public void testCliOverwriteConfigFile() throws Exception {
         new YAMLMapper().writeValue(file, testSinkConfig);
 
         Assert.assertEquals(testSinkConfig, CmdUtils.loadConfig(file.getAbsolutePath(), SinkConfig.class));
-        
+
         testMixCliAndConfigFile(
                 TENANT,
                 NAMESPACE,
@@ -866,7 +866,7 @@ public void testCliOverwriteConfigFile() throws Exception {
                 expectedSinkConfig
         );
     }
-    
+
     public void testMixCliAndConfigFile(
             String tenant,
             String namespace,
@@ -885,8 +885,8 @@ public void testMixCliAndConfigFile(
             String sinkConfigFile,
             SinkConfig sinkConfig
     ) throws Exception {
-        
-        
+
+
         // test create sink
         createSink.tenant = tenant;
         createSink.namespace = namespace;

pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Record.java → pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java

@@ -16,14 +16,12 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.io.core;
+package org.apache.pulsar.functions.api;
 
 import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
 
-import org.apache.pulsar.common.api.EncryptionContext;
-
 /**
  * Pulsar Connect's Record interface. Record encapsulates the information about a record being read from a Source.
  */
@@ -59,15 +57,6 @@ default Optional<Long> getRecordSequence() {
         return Optional.empty();
     }
 
-    /**
-     * Retrieves encryption-context that is attached to record.
-     *
-     * @return {@link Optional}<{@link EncryptionContext}>
-     */
-    default public Optional<EncryptionContext> getEncryptionCtx() {
-        return Optional.empty();
-    }
-
     /**
      * Retrieves user-defined properties attached to record.
      *

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java

@@ -53,6 +53,7 @@
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.functions.api.Function;
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.functions.proto.InstanceCommunication;
 import org.apache.pulsar.functions.proto.InstanceCommunication.MetricsData.Builder;
 import org.apache.pulsar.functions.proto.Function.SourceSpec;
@@ -67,7 +68,6 @@
 import org.apache.pulsar.functions.instance.state.StateContextImpl;
 import org.apache.pulsar.functions.utils.FunctionDetailsUtils;
 import org.apache.pulsar.functions.utils.Reflections;
-import org.apache.pulsar.io.core.Record;
 import org.apache.pulsar.io.core.Sink;
 import org.apache.pulsar.io.core.Source;
 import org.slf4j.Logger;

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/SinkRecord.java

@@ -24,8 +24,7 @@
 import lombok.AllArgsConstructor;
 import lombok.Data;
 
-import org.apache.pulsar.common.api.EncryptionContext;
-import org.apache.pulsar.io.core.Record;
+import org.apache.pulsar.functions.api.Record;
 
 @Data
 @AllArgsConstructor
@@ -58,12 +57,7 @@ public Optional<Long> getRecordSequence() {
         return sourceRecord.getRecordSequence();
     }
 
-    @Override
-    public Optional<EncryptionContext> getEncryptionCtx() {
-        return sourceRecord.getEncryptionCtx();
-    }
-
-    @Override
+     @Override
     public Map<String, String> getProperties() {
         return sourceRecord.getProperties();
     }

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java

@@ -35,6 +35,7 @@
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.functions.api.SerDe;
 import org.apache.pulsar.functions.api.utils.DefaultSerDe;
 import org.apache.pulsar.functions.instance.InstanceUtils;
@@ -45,7 +46,6 @@
 import org.apache.pulsar.functions.source.PulsarRecord;
 import org.apache.pulsar.functions.utils.FunctionConfig;
 import org.apache.pulsar.functions.utils.Reflections;
-import org.apache.pulsar.io.core.Record;
 import org.apache.pulsar.io.core.Sink;
 import org.apache.pulsar.io.core.SinkContext;
 

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarRecord.java

@@ -30,13 +30,12 @@
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.common.api.EncryptionContext;
 import org.apache.pulsar.functions.utils.Utils;
-import org.apache.pulsar.io.core.Record;
 
 @Builder
 @Getter
 @ToString
 @EqualsAndHashCode
-public class PulsarRecord<T> implements Record<T> {
+public class PulsarRecord<T> implements RecordWithEncryptionContext<T> {
 
     private final String topicName;
     private final int partition;
@@ -67,11 +66,7 @@ public Optional<Long> getRecordSequence() {
         return Optional.of(Utils.getSequenceId(message.getMessageId()));
     }
 
-    /**
-     * Retrieves encryption-context that is attached to record.
-     *
-     * @return {@link Optional}<{@link EncryptionContext}>
-     */
+    @Override
     public Optional<EncryptionContext> getEncryptionCtx() {
         return message.getEncryptionCtx();
     }

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java

@@ -30,13 +30,13 @@
 import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.client.impl.TopicMessageIdImpl;
 import org.apache.pulsar.client.impl.TopicMessageImpl;
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.functions.api.SerDe;
 import org.apache.pulsar.functions.api.utils.DefaultSerDe;
 import org.apache.pulsar.functions.instance.InstanceUtils;
 import org.apache.pulsar.functions.utils.FunctionConfig;
 import org.apache.pulsar.functions.utils.Reflections;
 import org.apache.pulsar.functions.utils.Utils;
-import org.apache.pulsar.io.core.Record;
 import org.apache.pulsar.io.core.Source;
 import org.apache.pulsar.io.core.SourceContext;
 

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/RecordWithEncryptionContext.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.pulsar.functions.source;
+
+import java.util.Optional;
+
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.functions.api.Record;
+
+public interface RecordWithEncryptionContext<T> extends Record<T> {
+
+    /**
+     * Retrieves encryption-context that is attached to record.
+     *
+     * @return {@link Optional}<{@link EncryptionContext}>
+     */
+    Optional<EncryptionContext> getEncryptionCtx();
+}

pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2ResourceTest.java

@@ -54,6 +54,7 @@
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.functions.api.Context;
 import org.apache.pulsar.functions.api.Function;
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.functions.api.utils.DefaultSerDe;
 import org.apache.pulsar.functions.proto.Function.FunctionDetails;
 import org.apache.pulsar.functions.proto.Function.FunctionMetaData;
@@ -68,7 +69,6 @@
 import org.apache.pulsar.functions.worker.WorkerService;
 import org.apache.pulsar.functions.worker.request.RequestResult;
 import org.apache.pulsar.functions.worker.rest.api.FunctionsImpl;
-import org.apache.pulsar.io.core.Record;
 import org.apache.pulsar.io.core.Sink;
 import org.apache.pulsar.io.core.SinkContext;
 import org.glassfish.jersey.media.multipart.FormDataContentDisposition;

pulsar-io/aerospike/src/main/java/org/apache/pulsar/io/aerospike/AerospikeAbstractSink.java

@@ -36,8 +36,8 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingDeque;
 
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.KeyValue;
-import org.apache.pulsar.io.core.Record;
 import org.apache.pulsar.io.core.Sink;
 import org.apache.pulsar.io.core.SinkContext;
 import org.slf4j.Logger;

pulsar-io/aerospike/src/main/java/org/apache/pulsar/io/aerospike/AerospikeStringSink.java

@@ -19,8 +19,8 @@
 
 package org.apache.pulsar.io.aerospike;
 
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.KeyValue;
-import org.apache.pulsar.io.core.Record;
 
 /**
  * Aerospike sink that treats incoming messages on the input topic as Strings

pulsar-io/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java

@@ -30,8 +30,8 @@
 
 import java.util.Map;
 
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.KeyValue;
-import org.apache.pulsar.io.core.Record;
 import org.apache.pulsar.io.core.Sink;
 import org.apache.pulsar.io.core.SinkContext;
 

pulsar-io/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraStringSink.java

@@ -19,8 +19,8 @@
 
 package org.apache.pulsar.io.cassandra;
 
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.KeyValue;
-import org.apache.pulsar.io.core.Record;
 
 /**
  * Cassandra sink that treats incoming messages on the input topic as Strings

pulsar-io/core/pom.xml

@@ -38,16 +38,10 @@
 
     <dependency>
       <groupId>${project.groupId}</groupId>
-      <artifactId>pulsar-common</artifactId>
+      <artifactId>pulsar-functions-api</artifactId>
       <version>${project.version}</version>
-      <exclusions>
-        <exclusion>
-          <groupId>*</groupId>
-          <artifactId>*</artifactId>
-        </exclusion>
-      </exclusions>
     </dependency>
-    
+
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>protobuf-shaded</artifactId>

pulsar-io/core/src/main/java/org/apache/pulsar/io/core/PushSource.java

@@ -21,6 +21,8 @@
 import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
 
+import org.apache.pulsar.functions.api.Record;
+
 /**
  * Pulsar's Push Source interface. PushSource read data from
  * external sources(database changes, twitter firehose, etc)

pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Sink.java

@@ -20,6 +20,8 @@
 
 import java.util.Map;
 
+import org.apache.pulsar.functions.api.Record;
+
 /**
  * Generic sink interface users can implement to run Sink on top of Pulsar Functions
  */

pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Source.java

@@ -20,6 +20,8 @@
 
 import java.util.Map;
 
+import org.apache.pulsar.functions.api.Record;
+
 public interface Source<T> extends AutoCloseable {
     /**
      * Open connector with configuration

pulsar-io/core/src/test/java/org/apache/pulsar/io/core/SinkTest.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.io.core;
 
+import org.apache.pulsar.functions.api.Record;
 import org.testng.annotations.Test;
 
 import java.util.HashMap;

pulsar-io/core/src/test/java/org/apache/pulsar/io/core/SourceTest.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.io.core;
 
+import org.apache.pulsar.functions.api.Record;
 import org.testng.annotations.Test;
 
 import java.util.HashMap;

pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java

@@ -29,8 +29,8 @@
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.KeyValue;
-import org.apache.pulsar.io.core.Record;
 import org.apache.pulsar.io.core.Sink;
 import org.apache.pulsar.io.core.SinkContext;
 

pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java

@@ -25,8 +25,8 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.PushSource;
-import org.apache.pulsar.io.core.Record;
 import org.apache.pulsar.io.core.SourceContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaStringSink.java

@@ -19,8 +19,8 @@
 
 package org.apache.pulsar.io.kafka;
 
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.KeyValue;
-import org.apache.pulsar.io.core.Record;
 
 /**
  * Kafka sink that treats incoming messages on the input topic as Strings