You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by sc...@apache.org on 2025/06/27 07:55:14 UTC
(beam) branch master updated: [Java Harness] Improve heap dump support by configuring JVM to dump heap on OOM exceptions (#32953)
This is an automated email from the ASF dual-hosted git repository.
scwhittle pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 45cb5b8b329 [Java Harness] Improve heap dump support by configuring JVM to dump heap on OOM exceptions (#32953)
45cb5b8b329 is described below
commit 45cb5b8b3294ad3dda15434319c0c2de49350170
Author: Sam Whittle <sc...@users.noreply.github.com>
AuthorDate: Fri Jun 27 09:55:06 2025 +0200
[Java Harness] Improve heap dump support by configuring JVM to dump heap on OOM exceptions (#32953)
* [Java Harness] Improve heap dump support by configuring JVM to dump heap on OOM exceptions to directory that the MemoryMonitor will look in.
Allow configuring remote upload location separately from tempLocation and add default-on support to gzip uploads to speed them up.
---
.../dataflow/worker/util/MemoryMonitorTest.java | 58 ++++--
sdks/java/container/boot.go | 12 ++
.../beam/sdk/options/MemoryMonitorOptions.java | 21 +++
.../beam/fn/harness/status/MemoryMonitor.java | 125 +++++++++----
.../beam/fn/harness/status/MemoryMonitorTest.java | 196 ++++++++++++++++++---
5 files changed, 340 insertions(+), 72 deletions(-)
diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/util/MemoryMonitorTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/util/MemoryMonitorTest.java
index 94800cd72f5..398d3f1e1c0 100644
--- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/util/MemoryMonitorTest.java
+++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/util/MemoryMonitorTest.java
@@ -65,15 +65,17 @@ public class MemoryMonitorTest {
private FakeGCStatsProvider provider;
private File localDumpFolder;
- private MemoryMonitor monitor;
- private Thread thread;
@Before
public void setup() throws IOException {
provider = new FakeGCStatsProvider();
localDumpFolder = tempFolder.newFolder();
+ }
+
+ @Test(timeout = 1000)
+ public void detectGCThrashing() throws InterruptedException {
// Update every 10ms, never shutdown VM.
- monitor =
+ MemoryMonitor monitor =
MemoryMonitor.forTest(
provider,
10,
@@ -85,12 +87,8 @@ public class MemoryMonitorTest {
"test-worker",
null,
Clock.systemUTC());
- thread = new Thread(monitor);
+ Thread thread = new Thread(monitor);
thread.start();
- }
-
- @Test(timeout = 1000)
- public void detectGCThrashing() throws InterruptedException {
monitor.waitForRunning();
monitor.waitForResources("Test1");
provider.inGCThrashingState.set(true);
@@ -107,16 +105,28 @@ public class MemoryMonitorTest {
monitor.waitForThrashingState(false);
assertTrue(s.tryAcquire(100, TimeUnit.MILLISECONDS));
monitor.waitForResources("Test3");
+ monitor.stop();
+ thread.join();
}
@Test
public void heapDumpOnce() throws Exception {
- File folder = tempFolder.newFolder();
-
- File dump1 = MemoryMonitor.dumpHeap(folder);
+ MemoryMonitor monitor =
+ MemoryMonitor.forTest(
+ provider,
+ 10,
+ 0,
+ false,
+ 50.0,
+ null,
+ localDumpFolder,
+ "test-worker",
+ null,
+ Clock.systemUTC());
+ File dump1 = monitor.dumpHeap();
assertNotNull(dump1);
assertTrue(dump1.exists());
- assertThat(dump1.getParentFile(), Matchers.equalTo(folder));
+ assertThat(dump1.getParentFile(), Matchers.equalTo(localDumpFolder));
}
@Test
@@ -137,7 +147,7 @@ public class MemoryMonitorTest {
@Test
public void uploadToGcs() throws Exception {
File remoteFolder = tempFolder.newFolder();
- monitor =
+ MemoryMonitor monitor =
MemoryMonitor.forTest(
provider,
10,
@@ -165,7 +175,7 @@ public class MemoryMonitorTest {
@Test
public void uploadJfrProfilesOnThrashing() throws Exception {
File remoteFolder = tempFolder.newFolder();
- monitor =
+ MemoryMonitor monitor =
MemoryMonitor.forTest(
provider,
10,
@@ -188,7 +198,7 @@ public class MemoryMonitorTest {
@Test
public void uploadToGcsDisabled() throws Exception {
- monitor =
+ MemoryMonitor monitor =
MemoryMonitor.forTest(
provider,
10,
@@ -210,6 +220,22 @@ public class MemoryMonitorTest {
@Test
public void disableMemoryMonitor() throws Exception {
+ // Update every 10ms, never shutdown VM.
+ MemoryMonitor monitor =
+ MemoryMonitor.forTest(
+ provider,
+ 10,
+ 0,
+ false,
+ 50.0,
+ null,
+ localDumpFolder,
+ "test-worker",
+ null,
+ Clock.systemUTC());
+ Thread thread = new Thread(monitor);
+ thread.start();
+
MemoryMonitor disabledMonitor =
MemoryMonitor.forTest(
provider,
@@ -232,5 +258,7 @@ public class MemoryMonitorTest {
// Enabled monitor thread should still be running.
assertTrue(thread.isAlive());
+ monitor.stop();
+ thread.join();
}
}
diff --git a/sdks/java/container/boot.go b/sdks/java/container/boot.go
index 7169c2628de..1f574d251cb 100644
--- a/sdks/java/container/boot.go
+++ b/sdks/java/container/boot.go
@@ -222,6 +222,18 @@ func main() {
} else {
args = append(args, jammAgentArgs)
}
+
+ // If heap dumping is enabled, configure the JVM to dump it on oom events.
+ if pipelineOptions, ok := info.GetPipelineOptions().GetFields()["options"]; ok {
+ if heapDumpOption, ok := pipelineOptions.GetStructValue().GetFields()["enableHeapDumps"]; ok {
+ if heapDumpOption.GetBoolValue() {
+ args = append(args, "-XX:+HeapDumpOnOutOfMemoryError",
+ "-Dbeam.fn.heap_dump_dir="+filepath.Join(dir, "heapdumps"),
+ "-XX:HeapDumpPath="+filepath.Join(dir, "heapdumps", "heap_dump.hprof"))
+ }
+ }
+ }
+
// Apply meta options
const metaDir = "/opt/apache/beam/options"
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/MemoryMonitorOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/MemoryMonitorOptions.java
index fc9bd3bc800..638916e1198 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/MemoryMonitorOptions.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/MemoryMonitorOptions.java
@@ -36,4 +36,25 @@ public interface MemoryMonitorOptions extends PipelineOptions {
Double getGCThrashingPercentagePerPeriod();
void setGCThrashingPercentagePerPeriod(Double value);
+
+ /**
+ * A remote file system to upload captured heap dumps to. If not provided defaults to
+ * --tempLocation.
+ *
+ * <p>This can be a path of any file system.
+ */
+ @Description("A remote file system to upload captured heap dumps to.")
+ String getRemoteHeapDumpLocation();
+
+ void setRemoteHeapDumpLocation(String value);
+
+ /**
+ * Controls if heap dumps that are copied to remote destination are gzipped compressed. This can
+ * greatly reduce the data copied and speed up the time to copy heap dumps.
+ */
+ @Description("A remote file system to upload captured heap dumps to.")
+ @Default.Boolean(true)
+ boolean getGzipCompressHeapDumps();
+
+ void setGzipCompressHeapDumps(boolean value);
}
diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/status/MemoryMonitor.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/status/MemoryMonitor.java
index cfd1ecb3cfa..9286c72faca 100644
--- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/status/MemoryMonitor.java
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/status/MemoryMonitor.java
@@ -21,6 +21,7 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
+import java.io.OutputStream;
import java.lang.management.GarbageCollectorMXBean;
import java.lang.management.ManagementFactory;
import java.nio.channels.Channels;
@@ -33,6 +34,7 @@ import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.zip.GZIPOutputStream;
import javax.management.InstanceNotFoundException;
import javax.management.MBeanException;
import javax.management.MBeanServer;
@@ -85,6 +87,8 @@ public class MemoryMonitor implements Runnable {
/** Amount of time (in ms) this thread must sleep between two consecutive iterations. */
public static final long DEFAULT_SLEEP_TIME_MILLIS = 15 * 1000; // 15 sec.
+ public static final String LOCAL_HEAPDUMP_DIR_SYSTEM_PROPERTY_NAME = "beam.fn.heap_dump_dir";
+
/**
* The number of periods to take into account when determining if the server is in GC thrashing.
*/
@@ -165,7 +169,7 @@ public class MemoryMonitor implements Runnable {
private byte[] reservedForDumpingHeap = new byte[HEAP_DUMP_RESERVED_BYTES];
/** If true, dump the heap when thrashing or requested. */
- private final boolean canDumpHeap;
+ @VisibleForTesting final boolean canDumpHeap;
/**
* The GC thrashing threshold for every period. If the time spent on garbage collection in one
@@ -190,17 +194,39 @@ public class MemoryMonitor implements Runnable {
/**
* If non null, if a heap dump is detected during initialization upload it to the given GCS path.
*/
- private final @Nullable String uploadFilePath;
+ @VisibleForTesting final @Nullable String uploadFilePath;
+
+ @VisibleForTesting final File localDumpFolder;
- private final File localDumpFolder;
+ @VisibleForTesting final boolean gzipCompress;
public static MemoryMonitor fromOptions(PipelineOptions options) {
SdkHarnessOptions sdkHarnessOptions = options.as(SdkHarnessOptions.class);
- String uploadFilePath = options.getTempLocation();
+ @Nullable String uploadFilePath = sdkHarnessOptions.getRemoteHeapDumpLocation();
+ if (uploadFilePath == null) {
+ uploadFilePath = options.getTempLocation();
+ }
PortablePipelineOptions portableOptions = options.as(PortablePipelineOptions.class);
- boolean canDumpHeap = uploadFilePath != null && portableOptions.getEnableHeapDumps();
- double gcThrashingPercentagePerPeriod = sdkHarnessOptions.getGCThrashingPercentagePerPeriod();
+ boolean canDumpHeap = false;
+ File localDumpFolder = getHeapDumpDir();
+ if (portableOptions.getEnableHeapDumps()) {
+ if (uploadFilePath == null) {
+ LOG.warn(
+ "Heap dumps are requested with --enableHeapDumps but neither --remoteHeapDumpLocation nor "
+ + "--tempLocation was provided to specify where to copy captured heap dumps to.");
+ } else {
+ try {
+ Files.createDirectories(localDumpFolder.toPath());
+ canDumpHeap = true;
+ } catch (Exception e) {
+ LOG.warn(
+ "Encountered exception creating directory for heap dumps, disabling heap dumping.",
+ e);
+ }
+ }
+ }
+ double gcThrashingPercentagePerPeriod = sdkHarnessOptions.getGCThrashingPercentagePerPeriod();
return new MemoryMonitor(
new SystemGCStatsProvider(),
DEFAULT_SLEEP_TIME_MILLIS,
@@ -208,7 +234,8 @@ public class MemoryMonitor implements Runnable {
canDumpHeap,
gcThrashingPercentagePerPeriod,
uploadFilePath,
- getLoggingDir());
+ getHeapDumpDir(),
+ sdkHarnessOptions.getGzipCompressHeapDumps());
}
@VisibleForTesting
@@ -219,7 +246,8 @@ public class MemoryMonitor implements Runnable {
boolean canDumpHeap,
double gcThrashingPercentagePerPeriod,
@Nullable String uploadFilePath,
- File localDumpFolder) {
+ File localDumpFolder,
+ boolean gzipCompress) {
return new MemoryMonitor(
gcStatsProvider,
sleepTimeMillis,
@@ -227,7 +255,8 @@ public class MemoryMonitor implements Runnable {
canDumpHeap,
gcThrashingPercentagePerPeriod,
uploadFilePath,
- localDumpFolder);
+ localDumpFolder,
+ gzipCompress);
}
private MemoryMonitor(
@@ -237,14 +266,16 @@ public class MemoryMonitor implements Runnable {
boolean canDumpHeap,
double gcThrashingPercentagePerPeriod,
@Nullable String uploadFilePath,
- File localDumpFolder) {
+ File localDumpFolder,
+ boolean gzipCompress) {
this.gcStatsProvider = gcStatsProvider;
this.sleepTimeMillis = sleepTimeMillis;
this.shutDownAfterNumGCThrashing = shutDownAfterNumGCThrashing;
- this.canDumpHeap = canDumpHeap;
this.gcThrashingPercentagePerPeriod = gcThrashingPercentagePerPeriod;
this.uploadFilePath = uploadFilePath;
this.localDumpFolder = localDumpFolder;
+ this.gzipCompress = gzipCompress;
+ this.canDumpHeap = canDumpHeap;
}
/** For testing only: Wait for the monitor to be running. */
@@ -297,42 +328,56 @@ public class MemoryMonitor implements Runnable {
@VisibleForTesting
boolean tryUploadHeapDumpIfItExists() {
- if (uploadFilePath == null) {
- return false;
- }
-
boolean uploadedHeapDump = false;
File localSource = getDefaultHeapDumpPath();
LOG.info("Looking for heap dump at {}", localSource);
if (localSource.exists()) {
- LOG.warn("Heap dump {} detected, attempting to upload file to ", localSource);
- String remoteDest =
- String.format("%s/heap_dump%s.hprof", uploadFilePath, UUID.randomUUID().toString());
- ResourceId resource = FileSystems.matchNewResource(remoteDest, false);
- try {
- uploadFile(localSource, resource);
- uploadedHeapDump = true;
- LOG.warn("Heap dump {} uploaded to {}", localSource, remoteDest);
- } catch (IOException e) {
- LOG.error("Error uploading heap dump to {}", remoteDest, e);
- }
+ if (uploadFilePath == null) {
+ LOG.warn("Heap dump {} detected but no upload directory was provided.", localSource);
+ } else {
+ String remoteDest =
+ String.format("%s/heap_dump%s.hprof", uploadFilePath, UUID.randomUUID());
+ if (gzipCompress) {
+ remoteDest = remoteDest + ".gz";
+ }
+ LOG.warn(
+ "Heap dump {} of {}MB detected, attempting to upload file to {}",
+ localSource,
+ localSource.length() / 1024 / 1024,
+ remoteDest);
+ ResourceId resource = FileSystems.matchNewResource(remoteDest, false);
+ try {
+ uploadFile(localSource, resource, gzipCompress);
+ uploadedHeapDump = true;
+ LOG.warn("Heap dump {} uploaded to {}", localSource, remoteDest);
+ } catch (IOException e) {
+ LOG.error("Error uploading heap dump to {}", remoteDest, e);
+ }
- try {
- Files.delete(localSource.toPath());
- LOG.info("Deleted local heap dump {}", localSource);
- } catch (IOException e) {
- LOG.warn("Unable to delete local heap dump {}", localSource, e);
+ try {
+ Files.delete(localSource.toPath());
+ LOG.info("Deleted local heap dump {}", localSource);
+ } catch (IOException e) {
+ LOG.warn("Unable to delete local heap dump {}", localSource, e);
+ }
}
}
return uploadedHeapDump;
}
- private void uploadFile(File srcPath, ResourceId destination) throws IOException {
+ private void uploadFile(File srcPath, ResourceId destination, boolean gzipCompress)
+ throws IOException {
StandardCreateOptions createOptions =
StandardCreateOptions.builder().setMimeType("application/octet-stream").build();
- try (WritableByteChannel dst = FileSystems.create(destination, createOptions)) {
- try (ReadableByteChannel src = Channels.newChannel(new FileInputStream(srcPath))) {
+ try (WritableByteChannel dst = FileSystems.create(destination, createOptions);
+ ReadableByteChannel src = Channels.newChannel(new FileInputStream(srcPath))) {
+ if (gzipCompress) {
+ try (OutputStream os = new GZIPOutputStream(Channels.newOutputStream(dst));
+ WritableByteChannel gzipDst = Channels.newChannel(os)) {
+ ByteStreams.copy(src, gzipDst);
+ }
+ } else {
ByteStreams.copy(src, dst);
}
}
@@ -560,7 +605,11 @@ public class MemoryMonitor implements Runnable {
}
/** Return the path for logging heap dumps. */
- private static File getLoggingDir() {
+ private static File getHeapDumpDir() {
+ @Nullable String heapDumpDir = System.getProperty(LOCAL_HEAPDUMP_DIR_SYSTEM_PROPERTY_NAME);
+ if (heapDumpDir != null) {
+ return new File(heapDumpDir);
+ }
return new File(System.getProperty("java.io.tmpdir"));
}
@@ -573,6 +622,8 @@ public class MemoryMonitor implements Runnable {
public File dumpHeap()
throws MalformedObjectNameException, InstanceNotFoundException, ReflectionException,
MBeanException, IOException {
+ Preconditions.checkState(
+ canDumpHeap, "Bug! Attempt to dump heap even though it should be disabled.");
return dumpHeap(localDumpFolder);
}
@@ -582,10 +633,10 @@ public class MemoryMonitor implements Runnable {
* <p>NOTE: We deliberately don't salt the heap dump filename so as to minimize disk impact of
* repeated dumps. These files can be of comparable size to the local disk.
*/
- @VisibleForTesting
- static synchronized File dumpHeap(File directory)
+ private static synchronized File dumpHeap(File directory)
throws MalformedObjectNameException, InstanceNotFoundException, ReflectionException,
MBeanException, IOException {
+
boolean liveObjectsOnly = false;
File fileName = new File(directory, "heap_dump.hprof");
if (fileName.exists() && !fileName.delete()) {
diff --git a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/status/MemoryMonitorTest.java b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/status/MemoryMonitorTest.java
index 2fa407cf9bf..f28df4f2d10 100644
--- a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/status/MemoryMonitorTest.java
+++ b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/status/MemoryMonitorTest.java
@@ -18,16 +18,22 @@
package org.apache.beam.fn.harness.status;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.io.File;
+import java.io.FileInputStream;
import java.io.IOException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.zip.GZIPInputStream;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.checkerframework.checker.nullness.qual.Nullable;
import org.hamcrest.Matchers;
+import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -61,21 +67,25 @@ public class MemoryMonitorTest {
private FakeGCStatsProvider provider;
private File localDumpFolder;
- private MemoryMonitor monitor;
- private Thread thread;
@Before
public void setup() throws IOException {
provider = new FakeGCStatsProvider();
localDumpFolder = tempFolder.newFolder();
- // Update every 10ms, never shutdown VM.
- monitor = MemoryMonitor.forTest(provider, 10, 0, false, 50.0, null, localDumpFolder);
- thread = new Thread(monitor);
- thread.start();
+ }
+
+ @After
+ public void tearDown() {
+ System.clearProperty(MemoryMonitor.LOCAL_HEAPDUMP_DIR_SYSTEM_PROPERTY_NAME);
}
@Test(timeout = 1000)
public void detectGCThrashing() throws InterruptedException {
+ // Update every 10ms, never shutdown VM.
+ MemoryMonitor monitor =
+ MemoryMonitor.forTest(provider, 10, 0, false, 50.0, null, localDumpFolder, false);
+ Thread thread = new Thread(monitor);
+ thread.start();
monitor.waitForRunning();
monitor.waitForResources("Test1");
provider.inGCThrashingState.set(true);
@@ -92,38 +102,41 @@ public class MemoryMonitorTest {
monitor.waitForThrashingState(false);
assertTrue(s.tryAcquire(100, TimeUnit.MILLISECONDS));
monitor.waitForResources("Test3");
+ monitor.stop();
+ thread.join();
}
@Test
public void heapDumpOnce() throws Exception {
- File folder = tempFolder.newFolder();
-
- File dump1 = MemoryMonitor.dumpHeap(folder);
+ MemoryMonitor monitor =
+ MemoryMonitor.forTest(provider, 10, 0, true, 50.0, null, localDumpFolder, false);
+ File dump1 = monitor.dumpHeap();
assertNotNull(dump1);
assertTrue(dump1.exists());
- assertThat(dump1.getParentFile(), Matchers.equalTo(folder));
+ assertThat(dump1.getParentFile(), Matchers.equalTo(localDumpFolder));
}
@Test
public void heapDumpTwice() throws Exception {
- File folder = tempFolder.newFolder();
-
- File dump1 = MemoryMonitor.dumpHeap(folder);
+ MemoryMonitor monitor =
+ MemoryMonitor.forTest(provider, 10, 0, true, 50.0, null, localDumpFolder, false);
+ File dump1 = monitor.dumpHeap();
assertNotNull(dump1);
assertTrue(dump1.exists());
- assertThat(dump1.getParentFile(), Matchers.equalTo(folder));
+ assertThat(dump1.getParentFile(), Matchers.equalTo(localDumpFolder));
- File dump2 = MemoryMonitor.dumpHeap(folder);
+ File dump2 = monitor.dumpHeap();
assertNotNull(dump2);
assertTrue(dump2.exists());
- assertThat(dump2.getParentFile(), Matchers.equalTo(folder));
+ assertThat(dump2.getParentFile(), Matchers.equalTo(localDumpFolder));
}
@Test
public void uploadFile() throws Exception {
File remoteFolder = tempFolder.newFolder();
- monitor =
- MemoryMonitor.forTest(provider, 10, 0, true, 50.0, remoteFolder.getPath(), localDumpFolder);
+ MemoryMonitor monitor =
+ MemoryMonitor.forTest(
+ provider, 10, 0, true, 50.0, remoteFolder.getPath(), localDumpFolder, false);
// Force the monitor to generate a local heap dump
monitor.dumpHeap();
@@ -137,9 +150,34 @@ public class MemoryMonitorTest {
assertThat(files[0].getAbsolutePath(), Matchers.containsString("hprof"));
}
+ @Test
+ public void uploadFileGzip() throws Exception {
+ File remoteFolder = tempFolder.newFolder();
+ MemoryMonitor monitor =
+ MemoryMonitor.forTest(
+ provider, 10, 0, true, 50.0, remoteFolder.getPath(), localDumpFolder, true);
+
+ // Force the monitor to generate a local heap dump
+ monitor.dumpHeap();
+
+ // Try to upload the heap dump
+ assertTrue(monitor.tryUploadHeapDumpIfItExists());
+
+ File[] files = remoteFolder.listFiles();
+ assertThat(files, Matchers.arrayWithSize(1));
+ assertThat(files[0].getAbsolutePath(), Matchers.containsString("heap_dump"));
+ assertThat(files[0].getAbsolutePath(), Matchers.containsString("hprof.gz"));
+
+ try (FileInputStream f = new FileInputStream(files[0]);
+ GZIPInputStream g = new GZIPInputStream(f)) {
+ assertThat(g.read(), Matchers.not(Matchers.is(-1)));
+ }
+ }
+
@Test
public void uploadFileDisabled() throws Exception {
- monitor = MemoryMonitor.forTest(provider, 10, 0, true, 50.0, null, localDumpFolder);
+ MemoryMonitor monitor =
+ MemoryMonitor.forTest(provider, 10, 0, true, 50.0, null, localDumpFolder, false);
// Force the monitor to generate a local heap dump
monitor.dumpHeap();
@@ -150,8 +188,14 @@ public class MemoryMonitorTest {
@Test
public void disableMemoryMonitor() throws Exception {
+ // Update every 10ms, never shutdown VM.
+ MemoryMonitor monitor =
+ MemoryMonitor.forTest(provider, 10, 0, false, 50.0, null, localDumpFolder, false);
+ Thread thread = new Thread(monitor);
+ thread.start();
+
MemoryMonitor disabledMonitor =
- MemoryMonitor.forTest(provider, 10, 0, true, 100.0, null, localDumpFolder);
+ MemoryMonitor.forTest(provider, 10, 0, true, 100.0, null, localDumpFolder, false);
Thread disabledMonitorThread = new Thread(disabledMonitor);
disabledMonitorThread.start();
@@ -162,5 +206,117 @@ public class MemoryMonitorTest {
// Enabled monitor thread should still be running.
assertTrue(thread.isAlive());
+ monitor.stop();
+ thread.join();
+ }
+
+ @Test
+ public void fromOptions_DefaultOff() throws Exception {
+ MemoryMonitor m = MemoryMonitor.fromOptions(PipelineOptionsFactory.fromArgs("").create());
+ assertFalse(m.canDumpHeap);
+ }
+
+ @Test
+ public void fromOptions_NoUploadLocationOff() throws Exception {
+ // No remote location specified.
+ MemoryMonitor m =
+ MemoryMonitor.fromOptions(PipelineOptionsFactory.fromArgs("--enableHeapDumps").create());
+ assertFalse(m.canDumpHeap);
+ }
+
+ @Test
+ public void fromOptions_EnabledAndUploadLocation() throws Exception {
+ @Nullable String ioLocalTempDir = System.getProperty("java.io.tmpdir");
+ MemoryMonitor m =
+ MemoryMonitor.fromOptions(
+ PipelineOptionsFactory.fromArgs("--enableHeapDumps", "--tempLocation=gs://temp/heaps")
+ .create());
+ if (ioLocalTempDir != null) {
+ assertTrue(m.canDumpHeap);
+ assertEquals(
+ new File(ioLocalTempDir).getCanonicalPath(), m.localDumpFolder.getCanonicalPath());
+ } else {
+ // If the generic temp directory property isn't set it will be disabled.
+ assertFalse(m.canDumpHeap);
+ }
+ }
+
+ @Test
+ public void fromOptions_EnabledSpecificPropertySetNoRemote() throws Exception {
+ System.setProperty(
+ MemoryMonitor.LOCAL_HEAPDUMP_DIR_SYSTEM_PROPERTY_NAME, localDumpFolder.toString());
+ MemoryMonitor m =
+ MemoryMonitor.fromOptions(PipelineOptionsFactory.fromArgs("--enableHeapDumps").create());
+ assertFalse(m.canDumpHeap);
+ }
+
+ @Test
+ public void fromOptions_SuccessfullyEnabledWithTemp() throws Exception {
+ System.setProperty(
+ MemoryMonitor.LOCAL_HEAPDUMP_DIR_SYSTEM_PROPERTY_NAME, localDumpFolder.toString());
+ MemoryMonitor m =
+ MemoryMonitor.fromOptions(
+ PipelineOptionsFactory.fromArgs("--enableHeapDumps", "--tempLocation=gs://temp/heaps")
+ .create());
+ assertTrue(m.canDumpHeap);
+ assertEquals(localDumpFolder, m.localDumpFolder);
+ assertEquals("gs://temp/heaps", m.uploadFilePath);
+ assertTrue(m.gzipCompress);
+ }
+
+ @Test
+ public void fromOptions_SuccessfullyEnabledWithSpecific() throws Exception {
+ System.setProperty(
+ MemoryMonitor.LOCAL_HEAPDUMP_DIR_SYSTEM_PROPERTY_NAME, localDumpFolder.toString());
+ MemoryMonitor m =
+ MemoryMonitor.fromOptions(
+ PipelineOptionsFactory.fromArgs(
+ "--enableHeapDumps", "--remoteHeapDumpLocation=gs://temp/heaps")
+ .create());
+ assertTrue(m.canDumpHeap);
+ assertEquals(localDumpFolder, m.localDumpFolder);
+ assertEquals("gs://temp/heaps", m.uploadFilePath);
+ assertTrue(m.gzipCompress);
+ }
+
+ @Test
+ public void fromOptions_SuccessfullyEnabledDisableGzip() throws Exception {
+ System.setProperty(
+ MemoryMonitor.LOCAL_HEAPDUMP_DIR_SYSTEM_PROPERTY_NAME, localDumpFolder.toString());
+ MemoryMonitor m =
+ MemoryMonitor.fromOptions(
+ PipelineOptionsFactory.fromArgs(
+ "--enableHeapDumps",
+ "--tempLocation=gs://temp/heaps",
+ "--gzipCompressHeapDumps=false")
+ .create());
+ assertTrue(m.canDumpHeap);
+ assertEquals(localDumpFolder, m.localDumpFolder);
+ assertEquals("gs://temp/heaps", m.uploadFilePath);
+ assertFalse(m.gzipCompress);
+ }
+
+ @Test
+ public void fromOptions_heapDumpToNonexistentDir() throws Exception {
+ File subfolder = localDumpFolder.toPath().resolve("subfolder_for_dumps").toFile();
+ File remoteFolder = tempFolder.newFolder();
+ System.setProperty(MemoryMonitor.LOCAL_HEAPDUMP_DIR_SYSTEM_PROPERTY_NAME, subfolder.toString());
+ MemoryMonitor m =
+ MemoryMonitor.fromOptions(
+ PipelineOptionsFactory.fromArgs(
+ "--enableHeapDumps", "--remoteHeapDumpLocation=" + remoteFolder)
+ .create());
+ assertTrue(m.canDumpHeap);
+ assertEquals(subfolder, m.localDumpFolder);
+ assertEquals(remoteFolder.toString(), m.uploadFilePath);
+ assertTrue(m.gzipCompress);
+
+ m.dumpHeap();
+ // Try to upload the heap dump
+ assertTrue(m.tryUploadHeapDumpIfItExists());
+ File[] files = remoteFolder.listFiles();
+ assertThat(files, Matchers.arrayWithSize(1));
+ assertThat(files[0].getAbsolutePath(), Matchers.containsString("heap_dump"));
+ assertThat(files[0].getAbsolutePath(), Matchers.containsString("hprof"));
}
}