adding support for authority header in gRPC sink

build.gradle

@@ -3,7 +3,7 @@ buildscript {
         maven { url 'https://plugins.gradle.org/m2/' }
     }
     dependencies {
-        classpath 'com.google.protobuf:protobuf-gradle-plugin:0.8.17'
+        classpath 'com.google.protobuf:protobuf-gradle-plugin:0.9.4'
         classpath "org.jfrog.buildinfo:build-info-extractor-gradle:4.4.7"
         classpath "org.ajoberstar:gradle-git:1.6.0"
     }
@@ -14,7 +14,7 @@ plugins {
     id 'idea'
     id 'checkstyle'
     id 'jacoco'
-    id "com.google.protobuf" version "0.8.17"
+    id "com.google.protobuf" version "0.9.4"
     id 'nebula.ospackage' version '8.6.3'
     id 'io.franzbecker.gradle-lombok' version '1.14'
     id 'maven-publish'
@@ -33,7 +33,7 @@ lombok {
 }
 
 group 'com.gotocompany'
-version '0.9.4'
+version '0.10.1'
 
 def projName = "firehose"
 
@@ -58,8 +58,8 @@ private Properties loadEnv() {
 def mainClassName = "com.gotocompany.firehose.launch.Main"
 
 dependencies {
-    implementation group: 'com.google.protobuf', name: 'protobuf-java', version: '3.1.0'
-    implementation group: 'com.google.protobuf', name: 'protobuf-java-util', version: '3.1.0'
+    implementation group: 'com.google.protobuf', name: 'protobuf-java', version: '3.25.0'
+    implementation group: 'com.google.protobuf', name: 'protobuf-java-util', version: '3.25.0'
     implementation group: 'com.datadoghq', name: 'java-dogstatsd-client', version: '2.13.0'
     implementation group: 'org.apache.httpcomponents', name: 'httpclient', version: '4.5.2'
     implementation group: 'org.aeonbits.owner', name: 'owner', version: '1.0.9'
@@ -101,7 +101,7 @@ dependencies {
     implementation platform('com.google.cloud:libraries-bom:20.5.0')
     implementation 'com.google.cloud:google-cloud-storage:2.20.1'
     implementation 'org.apache.logging.log4j:log4j-core:2.20.0'
-    implementation group: 'com.gotocompany', name: 'depot', version: '0.7.4'
+    implementation group: 'com.gotocompany', name: 'depot', version: '0.8.0'
     implementation group: 'com.networknt', name: 'json-schema-validator', version: '1.0.59' exclude group: 'org.slf4j'
 
     testImplementation group: 'junit', name: 'junit', version: '4.11'
@@ -117,11 +117,11 @@ dependencies {
 protobuf {
     generatedFilesBaseDir = "$projectDir/src/generated"
     protoc {
-        artifact = "com.google.protobuf:protoc:3.1.0"
+        artifact = "com.google.protobuf:protoc:3.25.0"
     }
     plugins {
         grpc {
-            artifact = "io.grpc:protoc-gen-grpc-java:1.0.3"
+            artifact = "io.grpc:protoc-gen-grpc-java:1.59.0"
         }
     }
     generateProtoTasks {

src/main/java/com/gotocompany/firehose/config/GrpcSinkConfig.java

@@ -17,4 +17,6 @@ public interface GrpcSinkConfig extends AppConfig {
     @Config.Key("SINK_GRPC_RESPONSE_SCHEMA_PROTO_CLASS")
     String getSinkGrpcResponseSchemaProtoClass();
 
+    @Config.Key("SINK_GRPC_SERVICE_AUTHORITY")
+    String getSinkGrpcServiceAuthority();
 }

src/main/java/com/gotocompany/firehose/sink/grpc/GrpcSinkFactory.java

@@ -25,8 +25,8 @@ public class GrpcSinkFactory {
     public static AbstractSink create(Map<String, String> configuration, StatsDReporter statsDReporter, StencilClient stencilClient) {
         GrpcSinkConfig grpcConfig = ConfigFactory.create(GrpcSinkConfig.class, configuration);
         FirehoseInstrumentation firehoseInstrumentation = new FirehoseInstrumentation(statsDReporter, GrpcSinkFactory.class);
-        String grpcSinkConfig = String.format("\n\tService host: %s\n\tService port: %s\n\tMethod url: %s\n\tResponse proto schema: %s",
-                grpcConfig.getSinkGrpcServiceHost(), grpcConfig.getSinkGrpcServicePort(), grpcConfig.getSinkGrpcMethodUrl(), grpcConfig.getSinkGrpcResponseSchemaProtoClass());
+        String grpcSinkConfig = String.format("\n\tService host: %s\n\tService port: %s\n\tAuthority: %s\n\tMethod url: %s\n\tResponse proto schema: %s",
+                grpcConfig.getSinkGrpcServiceHost(), grpcConfig.getSinkGrpcServicePort(), grpcConfig.getSinkGrpcServiceAuthority(), grpcConfig.getSinkGrpcMethodUrl(), grpcConfig.getSinkGrpcResponseSchemaProtoClass());
         firehoseInstrumentation.logDebug(grpcSinkConfig);
 
         ManagedChannel managedChannel = ManagedChannelBuilder.forAddress(grpcConfig.getSinkGrpcServiceHost(), grpcConfig.getSinkGrpcServicePort()).usePlaintext().build();

src/main/java/com/gotocompany/firehose/sink/grpc/client/GrpcClient.java

@@ -56,13 +56,17 @@ public DynamicMessage execute(byte[] logMessage, Headers headers) {
 
             Channel decoratedChannel = ClientInterceptors.intercept(managedChannel,
                      MetadataUtils.newAttachHeadersInterceptor(metadata));
+            CallOptions co = CallOptions.DEFAULT;
+            if (grpcSinkConfig.getSinkGrpcServiceAuthority() != null && !grpcSinkConfig.getSinkGrpcServiceAuthority().isEmpty()) {
+                co = co.withAuthority(grpcSinkConfig.getSinkGrpcServiceAuthority());
+            }
             byte[] response = ClientCalls.blockingUnaryCall(
                     decoratedChannel,
                     MethodDescriptor.newBuilder(marshaller, marshaller)
                             .setType(MethodDescriptor.MethodType.UNARY)
                             .setFullMethodName(grpcSinkConfig.getSinkGrpcMethodUrl())
                             .build(),
-                    CallOptions.DEFAULT,
+                    co,
                     logMessage);
 
             dynamicMessage = stencilClient.parse(grpcSinkConfig.getSinkGrpcResponseSchemaProtoClass(), response);