[CDAP-21079] Implement cdap-messaging-spi extension for spanner - Part 1

cdap-common/src/main/java/io/cdap/cdap/common/conf/Constants.java

@@ -2155,6 +2155,8 @@ public static final class MessagingSystem {
 
     // TMS HBase table attribute that indicates the number of prefix bytes used for the row key
     public static final String HBASE_MESSAGING_TABLE_PREFIX_NUM_BYTES = "cdap.messaging.table.prefix.num.bytes";
+
+    public static final String SPANNER_EXTENSION_PROPERTY_PREFIX = "messaging.spanner.properties.";
   }
 
   /**

cdap-common/src/main/resources/cdap-default.xml

@@ -2661,6 +2661,30 @@
     </description>
   </property>
 
+  <property>
+    <name>messaging.spanner.properties.publish.batch.size</name>
+    <value>50</value>
+    <description>
+      The default max number of messages in each publish batch for spanner messaging service.
+    </description>
+  </property>
+
+  <property>
+    <name>messaging.spanner.properties.publish.batch.timeout.millis</name>
+    <value>50</value>
+    <description>
+      The default maximum waiting time before a batch is published by spanner messaging service.
+    </description>
+  </property>
+
+  <property>
+    <name>messaging.spanner.properties.publish.batch.poll.millis</name>
+    <value>5</value>
+    <description>
+      The default polling frequency while either max batch size or batch timeout criterion is met.
+    </description>
+  </property>
+
   <property>
     <!-- Use lower heap memory ratio for the messaging service, since it uses non-heap memory for the connections -->
     <name>messaging.twill.java.heap.memory.ratio</name>
@@ -5384,7 +5408,8 @@
     <value>true</value>
     <description>
       Whether user code isolation is enabled in task worker. When enabled, task workers will ensure
-      only 1 request is accepted at max and later after the task worker is restarted it can take next request.
+      only 1 request is accepted at max and later after the task worker is restarted it can take
+      next request.
     </description>
   </property>
 
@@ -6515,7 +6540,8 @@
     <name>auditlog.messaging.fetch.size</name>
     <value>20</value>
     <description>
-      Number of messages to fetch from messaging system for each batch to consume by Audit log consumer.
+      Number of messages to fetch from messaging system for each batch to consume by Audit log
+      consumer.
     </description>
   </property>
 

cdap-master/pom.xml

@@ -108,6 +108,11 @@
       <artifactId>cdap-tms</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>cdap-messaging-ext-spanner</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.apache.tephra</groupId>
       <artifactId>tephra-api</artifactId>
@@ -245,6 +250,7 @@
         <stage.storage.ext.dir>${stage.opt.dir}/ext/storageproviders</stage.storage.ext.dir>
         <stage.authenticators.ext.dir>${stage.opt.dir}/ext/authenticators</stage.authenticators.ext.dir>
         <stage.credential.provider.extensions.dir>${stage.opt.dir}/ext/credentialproviders</stage.credential.provider.extensions.dir>
+        <stage.messaging.ext.dir>${stage.opt.dir}/ext/messagingproviders</stage.messaging.ext.dir>
         <stage.metricswriters.ext.dir>${stage.opt.dir}/ext/metricswriters/google_cloud_monitoring_writer
         </stage.metricswriters.ext.dir>
         <stage.eventwriters.ext.dir>${stage.opt.dir}/ext/eventwriters/google_cloud_pubsub_writer
@@ -705,6 +711,28 @@
                 </configuration>
               </execution>
 
+              <!-- Copy messaging extensions -->
+              <execution>
+                <id>copy-messaging-ext-spanner</id>
+                <phase>process-resources</phase>
+                <goals>
+                  <goal>copy-resources</goal>
+                </goals>
+                <configuration combine.self="override">
+                  <outputDirectory>${stage.messaging.ext.dir}/gcp-spanner</outputDirectory>
+                  <resources>
+                    <resource>
+                      <directory>
+                        ${project.parent.basedir}/cdap-messaging-ext-spanner/target/libexec/
+                      </directory>
+                      <includes>
+                        <include>*.jar</include>
+                      </includes>
+                    </resource>
+                  </resources>
+                </configuration>
+              </execution>
+
               <!-- Copy remote authenticator extensions -->
               <execution>
                 <id>copy-authenticator-ext-gcp</id>
@@ -713,7 +741,8 @@
                   <goal>copy-resources</goal>
                 </goals>
                 <configuration combine.self="override">
-                  <outputDirectory>${stage.authenticators.ext.dir}/gcp-remote-authenticator</outputDirectory>
+                  <outputDirectory>${stage.authenticators.ext.dir}/gcp-remote-authenticator
+                  </outputDirectory>
                   <resources>
                     <resource>
                       <directory>

cdap-messaging-ext-spanner/pom.xml

@@ -0,0 +1,118 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Copyright © 2016-2018 Cask Data, Inc.
+  Licensed 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>io.cdap.cdap</groupId>
+    <artifactId>cdap</artifactId>
+    <version>6.11.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>cdap-messaging-ext-spanner</artifactId>
+  <name>CDAP Google Cloud Spanner Messaging Extension</name>
+  <packaging>jar</packaging>
+
+  <properties>
+    <guava.version>30.1.1-jre</guava.version>
+    <spanner.version>6.12.2</spanner.version>
+  </properties>
+
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>com.google.cloud</groupId>
+        <artifactId>libraries-bom</artifactId>
+        <version>22.0.0</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+
+  <dependencies>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>cdap-messaging-spi</artifactId>
+      <version>${project.version}</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.google.cloud</groupId>
+      <artifactId>google-cloud-spanner</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.google.api</groupId>
+      <artifactId>gax</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+    </dependency>
+  </dependencies>
+
+  <profiles>
+    <profile>
+      <id>dist</id>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-dependency-plugin</artifactId>
+            <version>2.8</version>
+            <executions>
+              <execution>
+                <id>copy-dependencies</id>
+                <phase>prepare-package</phase>
+                <goals>
+                  <goal>copy-dependencies</goal>
+                </goals>
+                <configuration combine.self="override">
+                  <outputDirectory>${project.build.directory}/libexec</outputDirectory>
+                  <overWriteReleases>false</overWriteReleases>
+                  <overWriteSnapshots>false</overWriteSnapshots>
+                  <overWriteIfNewer>true</overWriteIfNewer>
+                  <prependGroupId>true</prependGroupId>
+                  <silent>true</silent>
+                  <includeScope>runtime</includeScope>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-jar-plugin</artifactId>
+            <version>2.4</version>
+            <executions>
+              <execution>
+                <id>jar</id>
+                <phase>prepare-package</phase>
+                <configuration combine.self="override">
+                  <outputDirectory>${project.build.directory}/libexec</outputDirectory>
+                  <finalName>${project.groupId}.${project.build.finalName}</finalName>
+                </configuration>
+                <goals>
+                  <goal>jar</goal>
+                </goals>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
+</project>