Recreation of GRPC publisher/subscribers

kafka-connector/README.md

@@ -87,6 +87,8 @@ Connector supports the following configs:
 | cps.topic | String | REQUIRED (No default) | The topic in Cloud Pub/Sub to publish to, e.g. "foo" for topic "/projects/bar/topics/foo". |
 | cps.project | String | REQUIRED (No default) | The project in Cloud Pub/Sub containing the topic, e.g. "bar" from above. |
 | maxBufferSize | Integer | 100 | The maximum number of messages that can be received for the messages on a topic partition before publishing them to Cloud Pub/Sub. |
+| maxBufferBytes | Long | 10000000 | The maximum number of bytes that can be received for the messages on a topic partition before publishing them to Cloud Pub/Sub. |
+| maxDelayThresholdMs | Integer | 100 | The maximum amount of time to wait to reach maxBufferSize or maxBufferBytes before publishing outstanding messages to Cloud Pub/Sub. |
 
 #### Schema Support and Data Model
 

kafka-connector/pom.xml

@@ -21,18 +21,18 @@
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-all</artifactId>
-      <version>1.6.1</version>
+      <version>1.10.1</version>
     </dependency>
     <dependency>
       <groupId>io.netty</groupId>
       <artifactId>netty-tcnative-boringssl-static</artifactId>
-      <version>2.0.3.Final</version>
+      <version>2.0.8.Final</version>
       <classifier>${os.detected.classifier}</classifier>
     </dependency>
     <dependency>
       <groupId>com.google.apis</groupId>
       <artifactId>google-api-services-pubsub</artifactId>
-      <version>v1-rev360-1.23.0</version>
+      <version>v1-rev381-1.23.0</version>
       <exclusions>
         <exclusion>
           <groupId>com.google.guava</groupId>
@@ -41,19 +41,25 @@
       </exclusions>
     </dependency>
     <dependency>
-      <groupId>com.google.api.grpc</groupId>
-      <artifactId>grpc-google-pubsub-v1</artifactId>
-      <version>0.1.1</version>
+      <groupId>com.google.cloud</groupId>
+      <artifactId>google-cloud-pubsub</artifactId>
+      <version>0.42.1-beta</version>
+    </dependency>
+    <!-- https://mvnrepository.com/artifact/com.google.protobuf/protobuf-java -->
+    <dependency>
+        <groupId>com.google.protobuf</groupId>
+        <artifactId>protobuf-java</artifactId>
+        <version>3.5.1</version>
     </dependency>
     <dependency>
       <groupId>com.google.auth</groupId>
       <artifactId>google-auth-library-oauth2-http</artifactId>
-      <version>0.8.0</version>
+      <version>0.9.0</version>
     </dependency>
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>connect-api</artifactId>
-      <version>0.10.2.0</version>
+      <version>0.10.2.1</version>
       <scope>provided</scope>
     </dependency>
     <dependency>
@@ -106,7 +112,7 @@
               <artifactSet>
                 <excludes>
                   <exclude>com.fasterxml.jackson.core:jackson-core:jar:*</exclude>
-		  <exclude>com.google.guava:*</exclude>
+                  <exclude>com.google.guava:*</exclude>
                 </excludes>
               </artifactSet>
               <transformers>
@@ -136,9 +142,9 @@
             protobuf-java directly, you will be transitively depending on the
             protobuf-java version that grpc depends on.
           -->
-          <protocArtifact>com.google.protobuf:protoc:3.0.0-beta-2:exe:${os.detected.classifier}</protocArtifact>
+          <protocArtifact>com.google.protobuf:protoc:3.5.1:exe:${os.detected.classifier}</protocArtifact>
           <pluginId>grpc-java</pluginId>
-          <pluginArtifact>io.grpc:protoc-gen-grpc-java:1.6.1:exe:${os.detected.classifier}</pluginArtifact>
+          <pluginArtifact>io.grpc:protoc-gen-grpc-java:1.11.0:exe:${os.detected.classifier}</pluginArtifact>
           <protoSourceRoot>${project.basedir}/src/main/proto</protoSourceRoot>
         </configuration>
         <executions>

kafka-connector/src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkConnector.java

@@ -38,7 +38,11 @@ public class CloudPubSubSinkConnector extends SinkConnector {
   private static final Logger log = LoggerFactory.getLogger(CloudPubSubSinkConnector.class);
 
   public static final String MAX_BUFFER_SIZE_CONFIG = "maxBufferSize";
+  public static final String MAX_BUFFER_BYTES_CONFIG = "maxBufferBytes";
+  public static final String MAX_DELAY_THRESHOLD_MS = "delayThresholdMs";
   public static final int DEFAULT_MAX_BUFFER_SIZE = 100;
+  public static final long DEFAULT_MAX_BUFFER_BYTES = 10000000L;
+  public static final int DEFAULT_DELAY_THRESHOLD_MS = 100;
   public static final String CPS_MESSAGE_BODY_NAME = "messageBodyName";
   public static final String DEFAULT_MESSAGE_BODY_NAME = "cps_message_body";
   public static final String PUBLISH_KAFKA_METADATA = "metadata.publish";
@@ -92,6 +96,22 @@ public ConfigDef config() {
             Importance.MEDIUM,
             "The maximum number of messages that can be received for the messages on a topic "
                 + "partition before publishing them to Cloud Pub/Sub.")
+        .define(
+            MAX_BUFFER_BYTES_CONFIG,
+            Type.LONG,
+            DEFAULT_MAX_BUFFER_BYTES,
+            ConfigDef.Range.between(1, DEFAULT_MAX_BUFFER_BYTES),
+            Importance.MEDIUM,
+            "The maximum number of bytes that can be received for the messages on a topic "
+                + "partition before publishing the messages to Cloud Pub/Sub.")
+        .define(
+            MAX_DELAY_THRESHOLD_MS,
+            Type.INT,
+            DEFAULT_DELAY_THRESHOLD_MS,
+            ConfigDef.Range.between(1, Integer.MAX_VALUE),
+            Importance.MEDIUM,
+            "The maximum amount of time to wait after receiving the first message in a batch for a "
+                + "before publishing the messages to Cloud Pub/Sub.")
         .define(
             PUBLISH_KAFKA_METADATA,
             Type.BOOLEAN,