* log-processor: removed log forwarding

Signed-off-by: neo <1100909+neowu@users.noreply.github.com>

Author: neowu

CHANGELOG.md

@@ -3,6 +3,8 @@
 ### 9.2.5 (8/27/2025 - )
 
 * db: check jdbc url on DBConfig
+* log-processor: removed log forwarding
+  > not really useful to mix business message with system message
 
 ### 9.2.4 (8/8/2025 - 8/21/2025)
 

ext/log-processor/conf/prod/resources/app.properties

@@ -5,4 +5,3 @@ app.index.shards=
 app.kibana.apiKey=
 app.kibana.banner=
 app.kibana.url=
-app.log.forward.config=

ext/log-processor/src/main/java/core/LogProcessorApp.java

@@ -1,16 +1,12 @@
 package core;
 
 import core.framework.http.HTTPClient;
-import core.framework.json.Bean;
-import core.framework.kafka.MessagePublisher;
 import core.framework.log.message.ActionLogMessage;
 import core.framework.log.message.EventMessage;
 import core.framework.log.message.LogTopics;
 import core.framework.log.message.StatMessage;
 import core.framework.module.App;
-import core.framework.module.KafkaConfig;
 import core.framework.search.module.SearchConfig;
-import core.log.LogForwardConfig;
 import core.log.domain.ActionDocument;
 import core.log.domain.EventDocument;
 import core.log.domain.StatDocument;
@@ -19,8 +15,6 @@
 import core.log.kafka.ActionLogMessageHandler;
 import core.log.kafka.EventMessageHandler;
 import core.log.kafka.StatMessageHandler;
-import core.log.service.ActionLogForwarder;
-import core.log.service.EventForwarder;
 import core.log.service.IndexOption;
 import core.log.service.IndexService;
 import core.log.service.JobConfig;
@@ -48,8 +42,7 @@ protected void initialize() {
 
         configureKibanaService();
 
-        Forwarders forwarders = configureLogForwarders();
-        configureKafka(forwarders);
+        configureKafka();
         configureJob();
 
         load(new DiagramModule());
@@ -73,15 +66,15 @@ private void configureIndexOption() {
         bind(option);
     }
 
-    private void configureKafka(Forwarders forwarders) {
+    private void configureKafka() {
         kafka().uri(requiredProperty("sys.kafka.uri"));
         kafka().concurrency(2);
         kafka().minPoll(1024 * 1024, Duration.ofMillis(500));          // try to get at least 1M message
         kafka().maxPoll(3000, 3 * 1024 * 1024);         // get 3M message at max
 
-        kafka().subscribe(LogTopics.TOPIC_ACTION_LOG, ActionLogMessage.class, bind(new ActionLogMessageHandler(forwarders.action)));
+        kafka().subscribe(LogTopics.TOPIC_ACTION_LOG, ActionLogMessage.class, bind(ActionLogMessageHandler.class));
         kafka().subscribe(LogTopics.TOPIC_STAT, StatMessage.class, bind(StatMessageHandler.class));
-        kafka().subscribe(LogTopics.TOPIC_EVENT, EventMessage.class, bind(new EventMessageHandler(forwarders.event)));
+        kafka().subscribe(LogTopics.TOPIC_EVENT, EventMessage.class, bind(EventMessageHandler.class));
     }
 
     private void configureSearch() {
@@ -109,30 +102,4 @@ private void configureJob() {
 
         schedule().dailyAt("cleanup-old-index-job", bind(CleanupOldIndexJob.class), LocalTime.of(1, 0));
     }
-
-    private Forwarders configureLogForwarders() {
-        var forwarders = new Forwarders();
-        // configure by env APP_LOG_FORWARD_CONFIG
-        String configValue = property("app.log.forward.config").orElse(null);
-        if (configValue != null) {
-            Bean.register(LogForwardConfig.class);
-            LogForwardConfig config = Bean.fromJSON(LogForwardConfig.class, configValue);
-            KafkaConfig kafka = kafka("forward");
-            kafka.uri(config.kafkaURI);
-            if (config.action != null) {
-                MessagePublisher<ActionLogMessage> publisher = kafka.publish(config.action.topic, ActionLogMessage.class);
-                forwarders.action = new ActionLogForwarder(publisher, config.action);
-            }
-            if (config.event != null) {
-                MessagePublisher<EventMessage> publisher = kafka.publish(config.event.topic, EventMessage.class);
-                forwarders.event = new EventForwarder(publisher, config.event);
-            }
-        }
-        return forwarders;
-    }
-
-    static class Forwarders {
-        ActionLogForwarder action;
-        EventForwarder event;
-    }
 }

ext/log-processor/src/main/java/core/log/kafka/ActionLogMessageHandler.java

@@ -9,9 +9,7 @@
 import core.framework.util.Maps;
 import core.log.domain.ActionDocument;
 import core.log.domain.TraceDocument;
-import core.log.service.ActionLogForwarder;
 import core.log.service.IndexService;
-import org.jspecify.annotations.Nullable;
 
 import java.time.LocalDate;
 import java.util.List;
@@ -21,25 +19,16 @@
  * @author neo
  */
 public class ActionLogMessageHandler implements BulkMessageHandler<ActionLogMessage> {
-    @Nullable
-    final ActionLogForwarder forwarder;
-
     @Inject
     IndexService indexService;
     @Inject
     ElasticSearchType<ActionDocument> actionType;
     @Inject
     ElasticSearchType<TraceDocument> traceType;
 
-    public ActionLogMessageHandler(@Nullable ActionLogForwarder forwarder) {
-        this.forwarder = forwarder;
-    }
-
     @Override
     public void handle(List<Message<ActionLogMessage>> messages) {
         index(messages, LocalDate.now());
-
-        if (forwarder != null) forwarder.forward(messages);
     }
 
     void index(List<Message<ActionLogMessage>> messages, LocalDate now) {

ext/log-processor/src/main/java/core/log/kafka/EventMessageHandler.java

@@ -8,9 +8,7 @@
 import core.framework.search.ElasticSearchType;
 import core.framework.util.Maps;
 import core.log.domain.EventDocument;
-import core.log.service.EventForwarder;
 import core.log.service.IndexService;
-import org.jspecify.annotations.Nullable;
 
 import java.time.LocalDate;
 import java.util.List;
@@ -20,23 +18,14 @@
  * @author neo
  */
 public class EventMessageHandler implements BulkMessageHandler<EventMessage> {
-    @Nullable
-    final EventForwarder forwarder;
-
     @Inject
     IndexService indexService;
     @Inject
     ElasticSearchType<EventDocument> eventType;
 
-    public EventMessageHandler(@Nullable EventForwarder forwarder) {
-        this.forwarder = forwarder;
-    }
-
     @Override
     public void handle(List<Message<EventMessage>> messages) {
         index(messages, LocalDate.now());
-
-        if (forwarder != null) forwarder.forward(messages);
     }
 
     void index(List<Message<EventMessage>> messages, LocalDate now) {

ext/log-processor/src/main/java/core/log/service/ActionLogForwarder.java

@@ -5,4 +5,3 @@ app.index.shards=1
 app.kibana.apiKey=
 app.kibana.banner=# local
 app.kibana.url=http://localhost:5601
-app.log.forward.config=