From 37ef06447aa6e8f96c6d9d765a684a7e82139b0c Mon Sep 17 00:00:00 2001 From: Ankit Kumar Date: Mon, 2 Mar 2026 22:11:11 +0530 Subject: [PATCH 1/6] Implement size-based file rotation for FileBasedEventLogger Signed-off-by: Ankit Kumar --- .../src/jvm/org/apache/storm/Config.java | 14 ++ .../storm/metric/FileBasedEventLogger.java | 76 +++++++-- .../metric/FileBasedEventLoggerTest.java | 145 ++++++++++++++++++ 3 files changed, 222 insertions(+), 13 deletions(-) create mode 100644 storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index a098e6cb632..e8d41260023 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -477,6 +477,20 @@ public class Config extends HashMap { @IsInteger @IsPositiveNumber(includeZero = true) public static final String TOPOLOGY_EVENTLOGGER_EXECUTORS = "topology.eventlogger.executors"; + /** + * The maximum size in MB for the event logger file before it rotates. + * If not specified, a default of 1000 MB is used. + */ + @IsInteger + @IsPositiveNumber + public static final String TOPOLOGY_EVENTLOGGER_ROTATION_SIZE_MB = "topology.eventlogger.rotation.size.mb"; + /** + * The maximum number of retained files for the event logger. + * If not specified, a default of 5 is used. + */ + @IsInteger + @IsPositiveNumber + public static final String TOPOLOGY_EVENTLOGGER_MAX_RETAINED_FILES = "topology.eventlogger.max.retained.files"; /** * The maximum amount of time given to the topology to fully process a message emitted by a spout. If the message is not acked within * this time frame, Storm will fail the message on the spout. Some spouts implementations will then replay the message at a later time. diff --git a/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java b/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java index 64ee85cd86c..0e6567b33ec 100644 --- a/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java +++ b/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java @@ -19,21 +19,23 @@ package org.apache.storm.metric; import java.io.BufferedWriter; -import java.io.File; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; import java.nio.file.StandardOpenOption; import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import org.apache.storm.Config; import org.apache.storm.shade.com.google.common.util.concurrent.ThreadFactoryBuilder; import org.apache.storm.task.TopologyContext; import org.apache.storm.utils.ConfigUtils; +import org.apache.storm.utils.ObjectReader; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,30 +43,39 @@ public class FileBasedEventLogger implements IEventLogger { private static final Logger LOG = LoggerFactory.getLogger(FileBasedEventLogger.class); private static final int FLUSH_INTERVAL_MILLIS = 1000; + private static final int DEFAULT_ROTATION_SIZE_MB = 100; + private static final int DEFAULT_MAX_RETAINED_FILES = 5; private Path eventLogPath; private BufferedWriter eventLogWriter; private ScheduledExecutorService flushScheduler; private volatile boolean dirty = false; + // File rotation configs + private long maxFileSize; + private int maxRetainedFiles; + private long currentFileSize = 0; + private void initLogWriter(Path logFilePath) { try { LOG.info("logFilePath {}", logFilePath); eventLogPath = logFilePath; + + currentFileSize = Files.exists(eventLogPath) ? Files.size(eventLogPath) : 0L; + eventLogWriter = Files.newBufferedWriter(eventLogPath, StandardCharsets.UTF_8, StandardOpenOption.CREATE, - StandardOpenOption.WRITE, StandardOpenOption.APPEND); + StandardOpenOption.WRITE, StandardOpenOption.APPEND); } catch (IOException e) { LOG.error("Error setting up FileBasedEventLogger.", e); throw new RuntimeException(e); } } - private void setUpFlushTask() { ThreadFactory threadFactory = new ThreadFactoryBuilder() - .setNameFormat("event-logger-flush-%d") - .setDaemon(true) - .build(); + .setNameFormat("event-logger-flush-%d") + .setDaemon(true) + .build(); flushScheduler = Executors.newSingleThreadScheduledExecutor(threadFactory); Runnable runnable = new Runnable() { @@ -82,15 +93,21 @@ public void run() { } }; - flushScheduler.scheduleAtFixedRate(runnable, FLUSH_INTERVAL_MILLIS, FLUSH_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); + flushScheduler.scheduleAtFixedRate(runnable, FLUSH_INTERVAL_MILLIS, FLUSH_INTERVAL_MILLIS, + TimeUnit.MILLISECONDS); } - @Override public void prepare(Map conf, Map arguments, TopologyContext context) { String stormId = context.getStormId(); int port = context.getThisWorkerPort(); + int rotationSizeMb = ObjectReader.getInt(conf.get(Config.TOPOLOGY_EVENTLOGGER_ROTATION_SIZE_MB), + DEFAULT_ROTATION_SIZE_MB); + this.maxFileSize = rotationSizeMb * 1024L * 1024L; + this.maxRetainedFiles = ObjectReader.getInt(conf.get(Config.TOPOLOGY_EVENTLOGGER_MAX_RETAINED_FILES), + DEFAULT_MAX_RETAINED_FILES); + /* * Include the topology name & worker port in the file name so that * multiple event loggers can log independently. @@ -98,9 +115,11 @@ public void prepare(Map conf, Map arguments, Top String workersArtifactRoot = ConfigUtils.workerArtifactsRoot(conf, stormId, port); Path path = Paths.get(workersArtifactRoot, "events.log"); - File dir = path.toFile().getParentFile(); - if (!dir.exists()) { - dir.mkdirs(); + try { + Files.createDirectories(path.getParent()); + } catch (IOException e) { + LOG.error("Failed to create directories for event logger", e); + throw new RuntimeException(e); } initLogWriter(path); setUpFlushTask(); @@ -109,9 +128,17 @@ public void prepare(Map conf, Map arguments, Top @Override public void log(EventInfo event) { try { - //TODO: file rotation - eventLogWriter.write(buildLogMessage(event)); + String logMessage = buildLogMessage(event); + byte[] logBytes = logMessage.getBytes(StandardCharsets.UTF_8); + int writeLength = logBytes.length + System.lineSeparator().length(); + + if (currentFileSize + writeLength > maxFileSize) { + rotateFiles(); + } + + eventLogWriter.write(logMessage); eventLogWriter.newLine(); + currentFileSize += writeLength; dirty = true; } catch (IOException ex) { LOG.error("Error logging event {}", event, ex); @@ -119,6 +146,29 @@ public void log(EventInfo event) { } } + private void rotateFiles() throws IOException { + eventLogWriter.close(); + + // Shift existing rotated files + for (int i = maxRetainedFiles - 1; i >= 1; i--) { + Path src = Paths.get(eventLogPath.toString() + "." + i); + Path dst = Paths.get(eventLogPath.toString() + "." + (i + 1)); + if (Files.exists(src)) { + Files.move(src, dst, StandardCopyOption.REPLACE_EXISTING); + } + } + + // Rename current events.log + if (Files.exists(eventLogPath)) { + Path dst = Paths.get(eventLogPath.toString() + ".1"); + Files.move(eventLogPath, dst, StandardCopyOption.REPLACE_EXISTING); + } + + // Re-open writers to empty file + initLogWriter(eventLogPath); + currentFileSize = 0; + } + protected String buildLogMessage(EventInfo event) { return event.toString(); } diff --git a/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java b/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java new file mode 100644 index 00000000000..fcc5abeceec --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.metric; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.storm.Config; +import org.apache.storm.task.TopologyContext; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class FileBasedEventLoggerTest { + + private Path tempDir; + private FileBasedEventLogger eventLogger; + + @BeforeEach + public void setUp() throws IOException { + tempDir = Files.createTempDirectory("storm-eventlogger-test"); + eventLogger = new FileBasedEventLogger(); + } + + @AfterEach + public void tearDown() throws IOException { + eventLogger.close(); + if (tempDir != null) { + Files.walk(tempDir) + .map(Path::toFile) + .forEach(File::delete); + tempDir.toFile().delete(); + } + } + + private TopologyContext mockTopologyContext() { + TopologyContext context = mock(TopologyContext.class); + when(context.getStormId()).thenReturn("test-topology-1"); + when(context.getThisWorkerPort()).thenReturn(6700); + return context; + } + + @Test + public void testFileRotation() throws IOException, InterruptedException { + Map conf = new HashMap<>(); + conf.put(Config.STORM_WORKERS_ARTIFACTS_DIR, tempDir.toAbsolutePath().toString()); + // We set rotation to be 1MB to trigger it easily, but we'll need to write + // a lot. Alternatively, we can use a very small value, but we need an int >= 1. + // Wait, Config is by MB. If we set it to 1, we still need to write 1MB. + // Let's reflection inject a smaller value for tests? No, Storm uses config. + // We will just use `1` MB and write a large string a few times. + conf.put(Config.TOPOLOGY_EVENTLOGGER_ROTATION_SIZE_MB, 1); + conf.put(Config.TOPOLOGY_EVENTLOGGER_MAX_RETAINED_FILES, 2); + + eventLogger.prepare(conf, new HashMap<>(), mockTopologyContext()); + + // 1 MB = 1048576 bytes + // We create an event message that is about 100KB, write it 11 times to exceed 1MB. + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < 100_000; i++) { + sb.append("A"); // 1 byte + } + String largeValue = sb.toString(); + + List values = new ArrayList<>(); + values.add(largeValue); + + // This toString() will add some bytes overhead, so each event is ~ 100KB. + IEventLogger.EventInfo eventInfo = new IEventLogger.EventInfo( + System.currentTimeMillis(), "test-component", 1, "msgId", values); + + // Write 10 times -> ~1 MB + for (int i = 0; i < 10; i++) { + eventLogger.log(eventInfo); + } + + // Wait a bit for flush if any (though rotation is synchronous in write) + Thread.sleep(100); + + Path expectedLogDir = tempDir.resolve("test-topology-1").resolve("6700"); + Path logFile = expectedLogDir.resolve("events.log"); + Path logFile1 = expectedLogDir.resolve("events.log.1"); + Path logFile2 = expectedLogDir.resolve("events.log.2"); + + // The first 10 writes should be in one file, almost 1 MB. + assertTrue(Files.exists(logFile)); + + // Write 2 more times to push it over 1MB + eventLogger.log(eventInfo); + eventLogger.log(eventInfo); + + Thread.sleep(100); + + // Now we expect events.log.1 to exist and events.log to be new + assertTrue(Files.exists(logFile1), "Rotated file events.log.1 should exist"); + + // Write 12 more times to push over 1MB again + for (int i = 0; i < 12; i++) { + eventLogger.log(eventInfo); + } + + Thread.sleep(100); + + // Now events.log.2 and events.log.1 and events.log should exist + assertTrue(Files.exists(logFile2), "Rotated file events.log.2 should exist"); + + // Write 12 MORE times to push over 1MB again + for (int i = 0; i < 12; i++) { + eventLogger.log(eventInfo); + } + + Thread.sleep(100); + + // max config was 2, so events.log.3 should NOT exist, and events.log.2 + // should exist. + Path logFile3 = expectedLogDir.resolve("events.log.3"); + assertTrue(!Files.exists(logFile3), "Rotated file events.log.3 should not exist"); + assertTrue(Files.exists(logFile2), "Rotated file events.log.2 should exist"); + } +} From 1244aabf779ba1f32da644d3a5811eb8c1304ff3 Mon Sep 17 00:00:00 2001 From: Ankit Kumar Date: Fri, 6 Mar 2026 15:33:20 +0530 Subject: [PATCH 2/6] Address review comments: use BYTES_PER_MB constant and commit writeLock changes --- .../storm/metric/FileBasedEventLogger.java | 42 ++++++++++++------- 1 file changed, 27 insertions(+), 15 deletions(-) diff --git a/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java b/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java index 0e6567b33ec..2766e156548 100644 --- a/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java +++ b/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java @@ -43,6 +43,7 @@ public class FileBasedEventLogger implements IEventLogger { private static final Logger LOG = LoggerFactory.getLogger(FileBasedEventLogger.class); private static final int FLUSH_INTERVAL_MILLIS = 1000; + private static final long BYTES_PER_MB = 1024L * 1024L; private static final int DEFAULT_ROTATION_SIZE_MB = 100; private static final int DEFAULT_MAX_RETAINED_FILES = 5; @@ -50,6 +51,7 @@ public class FileBasedEventLogger implements IEventLogger { private BufferedWriter eventLogWriter; private ScheduledExecutorService flushScheduler; private volatile boolean dirty = false; + private final Object writeLock = new Object(); // File rotation configs private long maxFileSize; @@ -82,9 +84,11 @@ private void setUpFlushTask() { @Override public void run() { try { - if (dirty) { - eventLogWriter.flush(); - dirty = false; + synchronized (writeLock) { + if (dirty && eventLogWriter != null) { + eventLogWriter.flush(); + dirty = false; + } } } catch (IOException ex) { LOG.error("Error flushing " + eventLogPath, ex); @@ -104,7 +108,7 @@ public void prepare(Map conf, Map arguments, Top int rotationSizeMb = ObjectReader.getInt(conf.get(Config.TOPOLOGY_EVENTLOGGER_ROTATION_SIZE_MB), DEFAULT_ROTATION_SIZE_MB); - this.maxFileSize = rotationSizeMb * 1024L * 1024L; + this.maxFileSize = rotationSizeMb * BYTES_PER_MB; this.maxRetainedFiles = ObjectReader.getInt(conf.get(Config.TOPOLOGY_EVENTLOGGER_MAX_RETAINED_FILES), DEFAULT_MAX_RETAINED_FILES); @@ -132,14 +136,18 @@ public void log(EventInfo event) { byte[] logBytes = logMessage.getBytes(StandardCharsets.UTF_8); int writeLength = logBytes.length + System.lineSeparator().length(); - if (currentFileSize + writeLength > maxFileSize) { - rotateFiles(); - } + synchronized (writeLock) { + if (currentFileSize + writeLength > maxFileSize) { + rotateFiles(); + } - eventLogWriter.write(logMessage); - eventLogWriter.newLine(); - currentFileSize += writeLength; - dirty = true; + if (eventLogWriter != null) { + eventLogWriter.write(logMessage); + eventLogWriter.newLine(); + currentFileSize += writeLength; + dirty = true; + } + } } catch (IOException ex) { LOG.error("Error logging event {}", event, ex); throw new RuntimeException(ex); @@ -175,14 +183,18 @@ protected String buildLogMessage(EventInfo event) { @Override public void close() { - try { - eventLogWriter.close(); + closeFlushScheduler(); + try { + synchronized (writeLock) { + if (eventLogWriter != null) { + eventLogWriter.close(); + eventLogWriter = null; + } + } } catch (IOException ex) { LOG.error("Error closing event log.", ex); } - - closeFlushScheduler(); } private void closeFlushScheduler() { From e8df12f875449ca3346d5efe8c523e62c450fe01 Mon Sep 17 00:00:00 2001 From: Richard Zowalla <13417392+rzo1@users.noreply.github.com> Date: Fri, 6 Mar 2026 11:12:43 +0100 Subject: [PATCH 3/6] Apply suggestion from @Copilot Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- storm-client/src/jvm/org/apache/storm/Config.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index e8d41260023..e332b726b28 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -479,7 +479,7 @@ public class Config extends HashMap { public static final String TOPOLOGY_EVENTLOGGER_EXECUTORS = "topology.eventlogger.executors"; /** * The maximum size in MB for the event logger file before it rotates. - * If not specified, a default of 1000 MB is used. + * If not specified, a default of 100 MB is used. */ @IsInteger @IsPositiveNumber From c6a148e1fe357357b6ab3aacacfde6a78a3558bd Mon Sep 17 00:00:00 2001 From: Richard Zowalla <13417392+rzo1@users.noreply.github.com> Date: Fri, 6 Mar 2026 11:12:59 +0100 Subject: [PATCH 4/6] Update storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java b/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java index fcc5abeceec..6248ddbd677 100644 --- a/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java +++ b/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java @@ -32,7 +32,6 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; From 936a6de19f16b3416068a48b5c8d070c61e15e14 Mon Sep 17 00:00:00 2001 From: Richard Zowalla Date: Fri, 6 Mar 2026 11:21:10 +0100 Subject: [PATCH 5/6] x --- .../jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java b/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java index 6248ddbd677..1db39b4ac5b 100644 --- a/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java +++ b/storm-client/test/jvm/org/apache/storm/metric/FileBasedEventLoggerTest.java @@ -31,7 +31,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; From 6fc4f3c095641a548363c771737333bf1a2c8c96 Mon Sep 17 00:00:00 2001 From: Ankit Kumar Date: Fri, 6 Mar 2026 16:26:12 +0530 Subject: [PATCH 6/6] Address remaining Copilot PR review comments --- .../storm/metric/FileBasedEventLogger.java | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java b/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java index 2766e156548..f626b839642 100644 --- a/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java +++ b/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java @@ -90,9 +90,8 @@ public void run() { dirty = false; } } - } catch (IOException ex) { + } catch (Exception ex) { LOG.error("Error flushing " + eventLogPath, ex); - throw new RuntimeException(ex); } } }; @@ -133,8 +132,7 @@ public void prepare(Map conf, Map arguments, Top public void log(EventInfo event) { try { String logMessage = buildLogMessage(event); - byte[] logBytes = logMessage.getBytes(StandardCharsets.UTF_8); - int writeLength = logBytes.length + System.lineSeparator().length(); + int writeLength = logMessage.length() + System.lineSeparator().length(); synchronized (writeLock) { if (currentFileSize + writeLength > maxFileSize) { @@ -157,8 +155,16 @@ public void log(EventInfo event) { private void rotateFiles() throws IOException { eventLogWriter.close(); + // Delete any files that exceed maxRetainedFiles (e.g. if the config was + // lowered) + int i = maxRetainedFiles; + while (Files.exists(Paths.get(eventLogPath.toString() + "." + i))) { + Files.delete(Paths.get(eventLogPath.toString() + "." + i)); + i++; + } + // Shift existing rotated files - for (int i = maxRetainedFiles - 1; i >= 1; i--) { + for (i = maxRetainedFiles - 1; i >= 1; i--) { Path src = Paths.get(eventLogPath.toString() + "." + i); Path dst = Paths.get(eventLogPath.toString() + "." + (i + 1)); if (Files.exists(src)) {