Pd 257044 databus changes rebase

queue/src/main/java/com/bazaarvoice/emodb/queue/QueueModule.java

@@ -34,6 +34,7 @@
 import com.google.inject.Provides;
 import com.google.inject.Singleton;
 import com.google.inject.TypeLiteral;
+import com.timgroup.statsd.StatsDClient;
 import org.apache.curator.framework.CuratorFramework;
 
 import java.time.Clock;
@@ -76,6 +77,7 @@ public QueueModule(MetricRegistry metricRegistry) {
     @Override
     protected void configure() {
         bind(CassandraFactory.class).asEagerSingleton();
+        bind(StatsDClient.class).asEagerSingleton();
 
         // Event Store
         bind(ChannelConfiguration.class).to(QueueChannelConfiguration.class).asEagerSingleton();

sor/src/main/java/com/bazaarvoice/emodb/sor/DataStoreModule.java

@@ -60,6 +60,7 @@
 import com.bazaarvoice.emodb.sor.db.cql.CqlForMultiGets;
 import com.bazaarvoice.emodb.sor.db.cql.CqlForScans;
 import com.bazaarvoice.emodb.sor.db.cql.SorCqlSettingsTask;
+import com.bazaarvoice.emodb.sor.kafka.KafkaProducerService;
 import com.bazaarvoice.emodb.sor.log.LogbackSlowQueryLogProvider;
 import com.bazaarvoice.emodb.sor.log.SlowQueryLog;
 import com.bazaarvoice.emodb.sor.log.SlowQueryLogConfiguration;
@@ -195,6 +196,7 @@ protected void configure() {
         bind(SlowQueryLog.class).toProvider(LogbackSlowQueryLogProvider.class);
         bind(HintsConsistencyTimeProvider.class).asEagerSingleton();
         bind(MinLagConsistencyTimeProvider.class).asEagerSingleton();
+        bind(KafkaProducerService.class);
 
         // The web servers are responsible for updating the ZooKeeper full consistency data.  CLI tools don't need to.
         // Enable updating the ZooKeeper full consistency data if specified

sor/src/main/java/com/bazaarvoice/emodb/sor/kafka/KafkaConfig.java

@@ -134,15 +134,6 @@ public static Properties getProducerProps() {
         return producerProps;
     }
 
-    // Kafka Admin properties
-    public static Properties getAdminProps() {
-        Properties adminProps = new Properties();
-
-        adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServersConfig);
-        logger.info("Kafka Admin properties initialized.");
-        return adminProps;
-    }
-
     // Ensure the SSM client is closed when the application shuts down
     public static void shutdown() {
         if (ssmClient != null) {