Support additional Kafka parameters for ftp and dcap.

This is an alternative to the configs property prefix for cells which are not implemented in terms of the spring framework.
indigo-dc · Oct 6, 2023 · c370ba2 · c370ba2
1 parent 60dd0ba
commit c370ba2
Show file tree

Hide file tree

Showing 7 changed files with 66 additions and 0 deletions.
diff --git a/modules/dcache-dcap/src/main/java/diskCacheV111/doors/DcapDoorSettings.java b/modules/dcache-dcap/src/main/java/diskCacheV111/doors/DcapDoorSettings.java
@@ -26,6 +26,9 @@
 import dmg.cells.nucleus.CellAddressCore;
 import dmg.cells.nucleus.CellEndpoint;
 import dmg.cells.nucleus.CellPath;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
@@ -42,8 +45,11 @@
 import org.dcache.poolmanager.PoolManagerStub;
 import org.dcache.services.login.RemoteLoginStrategy;
 import org.dcache.util.Option;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class DcapDoorSettings {
+    private static final Logger LOGGER = LoggerFactory.getLogger(DcapDoorSettings.class);
 
     @Option(name = "authorization")
     protected String auth;
@@ -99,6 +105,9 @@ public class DcapDoorSettings {
     @Option(name = "kafka-clientid")
     protected String kafkaclientid;
 
+    @Option(name = "kafka-config-file")
+    protected String kafkaConfigFile;
+
 
     @Option(name = "hsm",
           description = "Cell address of hsm manager",
@@ -332,6 +341,24 @@ public PoolManagerStub createPoolManagerStub(CellEndpoint cellEndpoint, CellAddr
 
     public KafkaProducer createKafkaProducer() {
         Properties props = new Properties();
+	if (kafkaConfigFile != null && !kafkaConfigFile.trim().isEmpty()) {
+	    try {
+		FileInputStream fis = new FileInputStream(kafkaConfigFile);
+		try {
+		    props.load(fis);
+		} catch (IOException ex) {
+		    LOGGER.error("failed to load configuration ", ex);
+		} finally {
+		    try {
+			fis.close();
+		    } catch (IOException ex) {
+			LOGGER.error("failed to close " + kafkaConfigFile, ex);
+		    }
+		}
+	    } catch (FileNotFoundException ex) {
+		LOGGER.error(ex.toString());
+	    }
+	}
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServer);
         props.put(ProducerConfig.CLIENT_ID_CONFIG, kafkaclientid);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,

diff --git a/modules/dcache-ftp/src/main/java/org/dcache/ftp/door/FtpDoorSettings.java b/modules/dcache-ftp/src/main/java/org/dcache/ftp/door/FtpDoorSettings.java
@@ -4,6 +4,9 @@
 import dmg.cells.nucleus.CellAddressCore;
 import dmg.cells.nucleus.CellEndpoint;
 import dmg.cells.nucleus.CellPath;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -14,13 +17,16 @@
 import org.dcache.util.Option;
 import org.dcache.util.OptionParser;
 import org.dcache.util.PortRange;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Object holding configuration options for FTP doors.
  * <p>
  * Settings can be injected using {@link OptionParser}.
  */
 public class FtpDoorSettings {
+    private static final Logger LOGGER = LoggerFactory.getLogger(FtpDoorSettings.class);
 
     @Option(name = "poolManager",
           description = "Well known name of the pool manager",
@@ -67,6 +73,9 @@ public class FtpDoorSettings {
     @Option(name = "kafka-clientid")
     protected String kafkaclientid;
 
+    @Option(name = "kafka-config-file")
+    protected String kafkaConfigFile;
+
 
     @Option(name = "clientDataPortRange",
           defaultValue = "0")
@@ -388,6 +397,24 @@ public PoolManagerStub createPoolManagerStub(CellEndpoint cellEndpoint,
 
     public KafkaProducer createKafkaProducer() {
         Properties props = new Properties();
+	if (kafkaConfigFile != null && !kafkaConfigFile.trim().isEmpty()) {
+	    try {
+		FileInputStream fis = new FileInputStream(kafkaConfigFile);
+		try {
+		    props.load(fis);
+		} catch (IOException ex) {
+		    LOGGER.error("failed to load configuration ", ex);
+		} finally {
+		    try {
+			fis.close();
+		    } catch (IOException ex) {
+			LOGGER.error("failed to close " + kafkaConfigFile, ex);
+		    }
+		}
+	    } catch (FileNotFoundException ex) {
+		LOGGER.error(ex.toString());
+	    }
+	}
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServer);
         props.put(ProducerConfig.CLIENT_ID_CONFIG, kafkaclientid);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,

diff --git a/skel/share/defaults/dcap.properties b/skel/share/defaults/dcap.properties
@@ -221,3 +221,6 @@ dcap.kafka.bootstrap-servers = ${dcache.kafka.bootstrap-servers}
 
 # Kafka topic name
 dcap.kafka.topic = ${dcache.kafka.topic}
+
+# File from which to load additional Kafka properties.
+dcap.kafka.config-file = ${dcache.kafka.config-file}
diff --git a/skel/share/defaults/ftp.properties b/skel/share/defaults/ftp.properties
@@ -396,3 +396,6 @@ ftp.kafka.bootstrap-servers = ${dcache.kafka.bootstrap-servers}
 
 # Kafka topic name
 ftp.kafka.topic = ${dcache.kafka.topic}
+
+# File from which to load additional Kafka properties.
+ftp.kafka.config-file = ${dcache.kafka.config-file}
diff --git a/skel/share/defaults/kafka.properties b/skel/share/defaults/kafka.properties
@@ -24,6 +24,10 @@ dcache.kafka.bootstrap-servers = localhost:9092
 # Kafka topic name
 dcache.kafka.topic = billing
 
+# Optional file from which to load additional Kafka properties for ftp and
+# dcap.  For other services, use dcache.kafka.configs.
+dcache.kafka.config-file =
+
 
 
 #--------------------Default Properties------------------------------

diff --git a/skel/share/services/dcap.batch b/skel/share/services/dcap.batch
@@ -119,6 +119,7 @@ create dmg.cells.services.login.LoginManager ${dcap.cell.name} \
              -kafka-max-block-units=${dcap.kafka.maximum-block.unit}\
              -retries-kafka=0 \
              -kafka-clientid=\"${dcap.cell.name}\" \
+             -kafka-config-file=\"${dcap.kafka.config-file}\" \
              -stageConfigurationFilePath=\"${dcap.authz.staging}\" \
              -allowAnonymousStaging=\"${dcap.authz.anonymous-staging}\" \
              -io-queue=${dcap.mover.queue} \

diff --git a/skel/share/services/ftp.batch b/skel/share/services/ftp.batch
@@ -94,6 +94,7 @@ create dmg.cells.services.login.LoginManager ${ftp.cell.name} \
    -billing=\"${ftp.service.billing}\" \
    -kafka=\"${ftp.enable.kafka}\" \
    -kafka-clientid=\"${ftp.cell.name}\" \
+   -kafka-config-file=\"${ftp.kafka.config-file}\" \
    -bootstrap-server-kafka=\"${ftp.kafka.bootstrap-servers}\" \
    -kafka-topic=\"${ftp.kafka.topic}\" \
    -kafka-max-block=${ftp.kafka.maximum-block}\