Merge branch 'master' into docubot-edfe9a5

Author: kofemann

docs/TheBook/src/main/markdown/kafkaproducer.md

@@ -51,15 +51,13 @@ Set kafka topic name
 dcache.kafka.topic = billing
 ```
 
-"billing" is default. The following service level variables
-reference dcache.kafka.topic:
+"billing" is default.
 
-   ftp.kafka.topic = ${dcache.kafka.topic}
-   pool.kafka.topic = ${dcache.kafka.topic}
-   webdav.kafka.topic = ${dcache.kafka.topic}
-   xrootd.kafka.topic = ${dcache.kafka.topic}
 
-> **Note**: Kafka producer support has been removed from the NFS door. The `nfs.kafka.*` properties are obsolete and no longer have any effect.
+**Note** that doors no longer supports pushing messages to Kafka; the
+producer for doors has been removed and any `nfs/ftp/webdav.kafka.*` properties are now
+obsolete.
+
 
 ### 3. Start Server
 

modules/dcache/src/main/java/diskCacheV111/services/TransferManager.java

@@ -5,7 +5,6 @@
 import diskCacheV111.util.CacheException;
 import diskCacheV111.util.MissingResourceCacheException;
 import diskCacheV111.util.PnfsId;
-import diskCacheV111.vehicles.DoorRequestInfoMessage;
 import diskCacheV111.vehicles.DoorTransferFinishedMessage;
 import diskCacheV111.vehicles.IpProtocolInfo;
 import diskCacheV111.vehicles.transferManager.CancelTransferMessage;
@@ -31,7 +30,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import org.dcache.cells.CellStub;
@@ -40,9 +38,7 @@
 import org.dcache.util.CDCExecutorServiceDecorator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.kafka.core.KafkaTemplate;
 
 
 /**
@@ -66,7 +62,6 @@ public abstract class TransferManager extends AbstractCellComponent
     private PoolManagerStub _poolManager;
     private CellStub _poolStub;
     private CellStub _billingStub;
-    private Consumer<DoorRequestInfoMessage> _kafkaSender = (s) -> {};
     private boolean _overwrite;
     private int _maxNumberOfDeleteRetries;
     // this is the timer which will timeout the
@@ -371,10 +366,6 @@ public CellStub getBillingStub() {
         return _billingStub;
     }
 
-    public Consumer<DoorRequestInfoMessage> getKafkaSender() {
-        return _kafkaSender;
-    }
-
     public String getIoQueueName() {
         return _ioQueueName;
     }
@@ -396,11 +387,6 @@ public void setBilling(CellStub billingStub) {
         _billingStub = billingStub;
     }
 
-    @Autowired(required = false)
-    public void setTransferTemplate(KafkaTemplate kafkaTemplate) {
-        _kafkaSender = kafkaTemplate::sendDefault;
-    }
-
     public void setPoolManager(PoolManagerStub poolManager) {
         _poolManager = poolManager;
     }

modules/dcache/src/main/java/diskCacheV111/services/TransferManagerHandler.java

@@ -6,7 +6,6 @@
 import static org.dcache.namespace.FileAttribute.STORAGECLASS;
 import static org.dcache.namespace.FileAttribute.STORAGEINFO;
 
-import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Sets;
 import diskCacheV111.util.CacheException;
@@ -61,7 +60,6 @@
 import org.dcache.vehicles.PnfsGetFileAttributes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.kafka.KafkaException;
 
 public class TransferManagerHandler extends AbstractMessageCallback<Message> {
 
@@ -742,13 +740,6 @@ void sendDoorRequestInfo(int code, String msg) {
         info.setResult(code, msg);
         LOGGER.debug("Sending info: {}", info);
         manager.getBillingStub().notify(info);
-
-        try {
-            manager.getKafkaSender().accept(info);
-        } catch (KafkaException | org.apache.kafka.common.KafkaException e) {
-            LOGGER.warn("Failed to send message to kafka: {} ",
-                        Throwables.getRootCause(e).getMessage());
-        }
     }
 
     public void timeout() {

modules/dcache/src/main/java/org/dcache/pool/classic/DefaultPostTransferService.java

@@ -20,7 +20,6 @@
 import static com.google.common.net.InetAddresses.toUriString;
 import static org.dcache.util.Exceptions.messageOrClassName;
 
-import com.google.common.base.Throwables;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import diskCacheV111.util.CacheException;
@@ -40,7 +39,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
 import org.dcache.cells.CellStub;
 import org.dcache.pool.movers.Mover;
 import org.dcache.pool.movers.TransferLifeCycle;
@@ -56,11 +54,7 @@
 import org.dcache.vehicles.FileAttributes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Required;
-import org.springframework.kafka.KafkaException;
-import org.springframework.kafka.core.KafkaTemplate;
 
 public class DefaultPostTransferService extends AbstractCellComponent implements
       PostTransferService, CellInfoProvider {
@@ -77,9 +71,6 @@ public class DefaultPostTransferService extends AbstractCellComponent implements
     private ChecksumModule _checksumModule;
     private CellStub _door;
 
-    private Consumer<MoverInfoMessage> _kafkaSender = (s) -> {
-    };
-
     private TransferLifeCycle transferLifeCycle;
 
     @Required
@@ -97,12 +88,6 @@ public void setChecksumModule(ChecksumModule checksumModule) {
         _checksumModule = checksumModule;
     }
 
-    @Autowired(required = false)
-    @Qualifier("transfer")
-    public void setKafkaTemplate(KafkaTemplate kafkaTemplate) {
-        _kafkaSender = kafkaTemplate::sendDefault;
-    }
-
     public void setTransferLifeCycle(TransferLifeCycle transferLifeCycle) {
         this.transferLifeCycle = transferLifeCycle;
     }
@@ -169,12 +154,6 @@ public void execute(final Mover<?> mover,
 
     private void sendBillingInfo(MoverInfoMessage moverInfoMessage) {
         _billing.notify(moverInfoMessage);
-
-        try {
-            _kafkaSender.accept(moverInfoMessage);
-        } catch (KafkaException | org.apache.kafka.common.KafkaException e) {
-            LOGGER.warn("Failed to send message to kafka: {} ", Throwables.getRootCause(e).getMessage());
-        }
     }
 
     public MoverInfoMessage generateBillingMessage(Mover<?> mover, long fileSize) {

modules/dcache/src/main/java/org/dcache/pool/classic/PoolV4.java

@@ -7,7 +7,6 @@
 import static java.util.stream.Collectors.toList;
 
 import com.google.common.base.Splitter;
-import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.net.InetAddresses;
@@ -87,7 +86,6 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ThreadFactory;
-import java.util.function.Consumer;
 import java.util.stream.Stream;
 import org.dcache.alarms.AlarmMarkerFactory;
 import org.dcache.alarms.PredefinedAlarm;
@@ -123,11 +121,7 @@
 import org.dcache.vehicles.FileAttributes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Required;
-import org.springframework.kafka.KafkaException;
-import org.springframework.kafka.core.KafkaTemplate;
 
 public class PoolV4
       extends AbstractCellComponent
@@ -209,9 +203,6 @@ public class PoolV4
 
     private Executor _executor;
 
-    private Consumer<RemoveFileInfoMessage> _kafkaSender = (s) -> {
-    };
-
     private ThreadFactory _threadFactory;
 
     // Hot file monitoring
@@ -239,12 +230,6 @@ protected void assertNotRunning(String error) {
         checkState(!_running, error);
     }
 
-    @Autowired(required = false)
-    @Qualifier("remove")
-    public void setKafkaTemplate(KafkaTemplate kafkaTemplate) {
-        _kafkaSender = kafkaTemplate::sendDefault;
-    }
-
     @Required
     public void setPoolName(String name) {
         assertNotRunning("Cannot change pool name after initialisation");
@@ -596,13 +581,6 @@ public void stateChanged(StateChangeEvent event) {
                 msg.setSubject(Subjects.ROOT);
                 msg.setResult(0, event.getWhy());
                 _billingStub.notify(msg);
-
-                try {
-                    _kafkaSender.accept(msg);
-                } catch (KafkaException | org.apache.kafka.common.KafkaException e) {
-                    LOGGER.warn("Failed to send message to kafka: {} ",
-                          Throwables.getRootCause(e).getMessage());
-                }
             }
         }
     }

modules/dcache/src/main/java/org/dcache/pool/nearline/NearlineStorageHandler.java

@@ -27,7 +27,6 @@
 import static org.dcache.util.Exceptions.messageOrClassName;
 
 import com.google.common.base.Functions;
-import com.google.common.base.Throwables;
 import com.google.common.util.concurrent.AsyncFunction;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -80,7 +79,6 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Consumer;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
@@ -119,11 +117,7 @@
 import org.dcache.vehicles.FileAttributes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Required;
-import org.springframework.kafka.KafkaException;
-import org.springframework.kafka.core.KafkaTemplate;
 
 /**
  * Entry point to and management interface for the nearline storage subsystem.
@@ -173,10 +167,6 @@ public record QueueStat(int queued, int active, int canceled)  {
 
     private CellAddressCore cellAddress;
 
-    private Consumer<StorageInfoMessage> _kafkaSender = (s) -> {
-    };
-
-
     @Override
     public void setCellAddress(CellAddressCore address) {
         cellAddress = address;
@@ -188,12 +178,6 @@ public void setScheduledExecutor(ScheduledExecutorService executor) {
     }
 
 
-    @Autowired(required = false)
-    @Qualifier("hsm")
-    public void setKafkaTemplate(KafkaTemplate kafkaTemplate) {
-        _kafkaSender = kafkaTemplate::sendDefault;
-    }
-
     @Required
     public void setExecutor(ListeningExecutorService executor) {
         this.executor = requireNonNull(executor);
@@ -1187,11 +1171,6 @@ private void done(@Nullable Throwable cause) {
             addFromNearlineStorage(infoMsg, storage);
 
             billingStub.notify(infoMsg);
-            try {
-                _kafkaSender.accept(infoMsg);
-            } catch (KafkaException | org.apache.kafka.common.KafkaException e) {
-                LOGGER.warn("Failed to send message to kafka: {} ", Throwables.getRootCause(e).getMessage());
-            }
             flushRequests.removeAndCallback(pnfsId, cause);
         }
 
@@ -1421,11 +1400,6 @@ private void done(@Nullable Throwable cause) {
             addFromNearlineStorage(infoMsg, storage);
 
             billingStub.notify(infoMsg);
-            try {
-                _kafkaSender.accept(infoMsg);
-            } catch (KafkaException | org.apache.kafka.common.KafkaException e) {
-                LOGGER.warn("Failed to send message to kafka: {} ", Throwables.getRootCause(e).getMessage());
-            }
             stageRequests.removeAndCallback(pnfsId, cause);
         }
 

modules/dcache/src/main/java/org/dcache/util/jetty/RateLimitedHandlerList.java

@@ -240,6 +240,10 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
             return;
         }
 
+
+
+        // REVISIT: do we want to keep per-client rate limiter or only for authentication errors?
+        /*
         if (!getClientRateLimiter(client).tryAcquire()) {
             LOGGER.debug("Blocking client with too many requests {}", client);
             response.setStatus(HttpStatus.TOO_MANY_REQUESTS_429);
@@ -257,7 +261,7 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
             baseRequest.setHandled(true);
             return;
         }
-
+         */
         Handler[] handlers = this.getHandlers();
         if (handlers != null && this.isStarted()) {
             for (Handler handler : handlers) {

modules/dcache/src/main/resources/diskCacheV111/services/transfer-manager.xml

@@ -82,33 +82,4 @@
         <property name="overwrite" value="false"/>
     </bean>
 
-  <beans profile="kafka-true">
-
-      <bean id="listener" class="org.dcache.kafka.LoggingProducerListener"/>
-
-      <bean id="kafka-configs"
-            class="org.dcache.util.configuration.ConfigurationMapFactoryBean">
-        <property name="prefix" value="transfermanagers.kafka.producer.configs"/>
-        <property name="staticEnvironment">
-            <map>
-                <entry key="bootstrap.servers" value="${transfermanagers.kafka.producer.bootstrap.servers}"/>
-                <entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
-                <entry key="value.serializer" value="org.dcache.notification.DoorRequestMessageSerializer"/>
-                <entry key="client.id" value="${transfermanagers.cell.name}@${dcache.domain.name}"/>
-            </map>
-        </property>
-      </bean>
-
-      <bean id="billing-template" class="org.springframework.kafka.core.KafkaTemplate">
-          <constructor-arg>
-              <bean class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
-                  <constructor-arg name="configs" ref="kafka-configs"/>
-              </bean>
-          </constructor-arg>
-          <property name="defaultTopic" value="${transfermanagers.kafka.topic}"/>
-          <property name="producerListener" ref="listener"/>
-      </bean>
-
-  </beans>
-
 </beans>