dcap: global kafka producer

Motivation

Currently the code is implemented in a way that each new transfer creates a new dcap client and for each of this clients a unique kafka procedure is created.
Apparently this is not the optimal way, and there should be created one (global) kafka producer when the door is started.

Modifications

the prducer is now created in  `init()` method and is latter accessed in each new instance of `DCapDoorInterpreterV3`.

Result

A global Producer is now created when the Dcap door is started and is closed when the door is stoped.

Target: master
Requires-notes: no
Requires-book: no
Avted-by: Tigran

modules/dcache-dcap/src/main/java/diskCacheV111/doors/DCapDoorInterpreterV3.java

@@ -233,12 +233,8 @@ public DCapDoorInterpreterV3(CellEndpoint cell, CellAddressCore address, DcapDoo
         _loginStrategy = settings.createLoginStrategy(cell);
 
         if ( _settings.isKafkaEnabled() ){
-            _kafkaProducer = settings.createKafkaProducer(_settings.getKafkaBootstrapServer(),
-                    address.toString(),
-                    _settings.getKafkaMaxBlockMs(),
-                    _settings.getKafkaRetries());
+            _kafkaProducer = settings.getKafkaProducer();
         }
-
         _startedTS = new Date();
     }
 
@@ -2471,18 +2467,6 @@ public void close() {
         }
     }
 
-    protected void shutdownKafka() {
-        /*The producer consists of a pool of buffer space that holds records that haven't yet been
-          transmitted to the server as well as a background I/O thread
-          that is responsible for turning these records into requests and transmitting them to the cluster.
-          Failure to close the producer after use will leak these resources. Hence we need to  close Kafka Producer
-         */
-        if (_settings.isKafkaEnabled()) {
-            _log.debug("Shutting down kafka");
-            _kafkaProducer.close();
-        }
-    }
-
     @Override
     public void   getInfo( PrintWriter pw ){
         pw.println( " ----- DCapDoorInterpreterV3 ----------" ) ;

modules/dcache-dcap/src/main/java/diskCacheV111/doors/DcapDoorSettings.java

@@ -166,6 +166,9 @@ public class DcapDoorSettings
 
     private CheckStagePermission checkStagePermission;
 
+    private KafkaProducer _kafkaProducer;
+
+
     public void init()
     {
         isAuthorizationStrong = (auth != null) && auth.equals("strong");
@@ -198,6 +201,10 @@ public void init()
 
         checkStagePermission = new CheckStagePermission(stageConfigurationFilePath);
         checkStagePermission.setAllowAnonymousStaging(allowAnonymousStaging);
+        if (isKafkaEnabled) {
+            _kafkaProducer = createKafkaProducer(kafkaBootstrapServer, String.valueOf(kafkaMaxBlock), kafkaRetries);
+            _log.warn("Creating KafkaProducer" + _kafkaProducer.hashCode());
+        }
     }
 
     public CellPath getPnfsManager()
@@ -221,17 +228,17 @@ public boolean isKafkaEnabled() {
         return isKafkaEnabled;
     }
 
-    /**
-     * Returns a list of host/port pairs (brokers) to use for establishing the initial connection to the Kafka cluster.
-     * This list is just used to discover the rest of the brokers in the cluster and should be in the form
-     * host1:port1,host2:port2,....
-     *
-     * @return    the list of  of host/port pairs
-     */
-    public String getKafkaBootstrapServer() {
-        return kafkaBootstrapServer;
+
+    public KafkaProducer getKafkaProducer() {
+        return _kafkaProducer;
     }
 
+    public void destroy() {
+        if (isKafkaEnabled){
+            _log.warn("Shutting down kafka");
+            _kafkaProducer.close();
+        }
+    }
     /**
      * Returns the name of kafka topic
      *
@@ -241,16 +248,6 @@ public String getKafkaTopic() {
         return kafkaTopic;
     }
 
-    /**
-     * Returns the parameter that controls how long
-     * how long the producer will block when calling send().
-     *
-     * @retrun a timeframe during which producer will block sending messages, by default set to 60000
-     */
-    public String getKafkaMaxBlockMs() {
-        return String.valueOf(TimeUnit.MILLISECONDS.convert(kafkaMaxBlock, kafkaMaxBlockUnits));
-    }
-
     /**
      * Returns the number of retries that the producer will retry sending the messages before failing it.
      *
@@ -344,15 +341,15 @@ public PoolManagerStub createPoolManagerStub(CellEndpoint cellEndpoint, CellAddr
         return stub;
     }
 
+
+
     public KafkaProducer createKafkaProducer(String bootstrap_server,
-                                             String client_id,
                                              String max_block_ms,
                                              String retries)
     {
         Properties props = new Properties();
-
+        //TODO check client_id
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap_server);
-        props.put(ProducerConfig.CLIENT_ID_CONFIG, client_id);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.dcache.notification.DoorRequestMessageSerializer");
         props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, max_block_ms);

modules/dcache-dcap/src/main/java/diskCacheV111/doors/DcapInterpreterFactory.java

@@ -21,7 +21,6 @@
 import java.util.concurrent.Executor;
 
 import diskCacheV111.util.ConfigurationException;
-
 import dmg.cells.nucleus.CellAddressCore;
 import dmg.cells.nucleus.CellEndpoint;
 import dmg.util.StreamEngine;
@@ -42,6 +41,7 @@ public class DcapInterpreterFactory implements LineBasedInterpreterFactory
     @Override
     public void configure(Args args) throws ConfigurationException
     {
+        //TODO this is called three times on start to be check
         new OptionParser(args).inject(settings).init();
     }
 
@@ -51,4 +51,10 @@ public LineBasedInterpreter create(CellEndpoint endpoint, CellAddressCore myAddr
     {
         return new DcapLineBasedInterpreterAdapter(endpoint, myAddress, engine, settings, poolManagerHandler);
     }
+
+    @Override
+    public void destroy()  {
+         settings.destroy();
+
+    }
 }

modules/dcache-dcap/src/main/java/diskCacheV111/doors/DcapLineBasedInterpreterAdapter.java

@@ -108,7 +108,7 @@ public void shutdown()
     @Override
     public void messagingClosed()
     {
-        shutdownKafka();
+        //TODO delete
     }
 
     @Override

modules/dcache/src/main/java/diskCacheV111/doors/LineBasedDoorFactory.java

@@ -95,6 +95,7 @@ protected void doStart()
     @Override
     protected void doStop()
     {
+        factory.destroy();
         poolManagerHandler.beforeStop();
         executor.shutdown();
         notifyStopped();

modules/dcache/src/main/java/diskCacheV111/doors/LineBasedInterpreter.java

@@ -55,6 +55,7 @@ public interface LineBasedInterpreter
      * the client has not yet been closed, but the client might not accept any
      * more data.
      */
+    //TODO not clear why do we need this method, kafka Producer is not closed anymore in this methods implementation
     default void messagingClosed()
     {
         // do nothing.

modules/dcache/src/main/java/diskCacheV111/doors/LineBasedInterpreterFactory.java

@@ -61,4 +61,11 @@ public interface LineBasedInterpreterFactory
      */
     LineBasedInterpreter create(CellEndpoint endpoint, CellAddressCore myAddress,
                                 StreamEngine engine, Executor executor, PoolManagerHandler poolManager) throws Exception;
+
+
+    /**
+    * Free allocated  resources
+     */
+    void destroy();
+
 }