diff --git a/apm-agent-common/src/main/java/co/elastic/apm/agent/common/util/AgentInfo.java b/apm-agent-common/src/main/java/co/elastic/apm/agent/common/util/AgentInfo.java
index b677ceb7c4..dbd2b1a9de 100644
--- a/apm-agent-common/src/main/java/co/elastic/apm/agent/common/util/AgentInfo.java
+++ b/apm-agent-common/src/main/java/co/elastic/apm/agent/common/util/AgentInfo.java
@@ -36,7 +36,8 @@ public class AgentInfo {
         "com.blogspot.mydailyjava.weaklockfree",
         "com.lmax.disruptor",
         "com.dslplatform.json",
-        "com.googlecode.concurrentlinkedhashmap"
+        "com.googlecode.concurrentlinkedhashmap",
+        "co.elastic.otel"
     ));
 
     private static final Set<String> agentRootPackages = new HashSet<>(Arrays.asList(
diff --git a/apm-agent-core/pom.xml b/apm-agent-core/pom.xml
index cb04140691..5f35f7912d 100644
--- a/apm-agent-core/pom.xml
+++ b/apm-agent-core/pom.xml
@@ -100,6 +100,11 @@
             <artifactId>HdrHistogram</artifactId>
             <version>2.1.11</version>
         </dependency>
+        <dependency>
+            <groupId>co.elastic.otel</groupId>
+            <artifactId>jvmti-access</artifactId>
+            <version>0.3.0</version>
+        </dependency>
         <!--
         We can't use caffeine due to requiring Java 7.
         As recommended by the author, we use concurrentlinkedhashmap-lru instead:
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/configuration/UniversalProfilingConfiguration.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/configuration/UniversalProfilingConfiguration.java
new file mode 100644
index 0000000000..98ca6b0cb5
--- /dev/null
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/configuration/UniversalProfilingConfiguration.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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 co.elastic.apm.agent.configuration;
+
+import org.stagemonitor.configuration.ConfigurationOption;
+import org.stagemonitor.configuration.ConfigurationOptionProvider;
+
+import static co.elastic.apm.agent.tracer.configuration.RangeValidator.isInRange;
+
+public class UniversalProfilingConfiguration extends ConfigurationOptionProvider {
+
+    private static final String PROFILING_CATEGORY = "Profiling";
+
+    private final ConfigurationOption<Boolean> enabled = ConfigurationOption.booleanOption()
+        .key("universal_profiling_integration_enabled")
+        .tags("added[1.50.0]", "internal")
+        .configurationCategory(PROFILING_CATEGORY)
+        .description("If enabled, the apm agent will correlate it's transaction with the profiling data from elastic universal profiling running on the same host.")
+        .buildWithDefault(false);
+
+    private final ConfigurationOption<Long> bufferSize = ConfigurationOption.longOption()
+        .key("universal_profiling_integration_buffer_size")
+        .addValidator(isInRange(64L, Long.MAX_VALUE))
+        .tags("added[1.50.0]", "internal")
+        .configurationCategory(PROFILING_CATEGORY)
+        .description("The feature needs to buffer ended local-root spans for a short duration to ensure that all of its profiling data has been received." +
+                     "This configuration option configures the buffer size in number of spans. " +
+                     "The higher the number of local root spans per second, the higher this buffer size should be set.\n" +
+                     "The agent will log a warning if it is not capable of buffering a span due to insufficient buffer size. " +
+                     "This will cause the span to be exported immediately instead with possibly incomplete profiling correlation data.")
+        .buildWithDefault(4096L);
+
+    private final ConfigurationOption<String> socketDir = ConfigurationOption.stringOption()
+        .key("universal_profiling_integration_socket_dir")
+        .tags("added[1.50.0]", "internal")
+        .configurationCategory(PROFILING_CATEGORY)
+        .description("The extension needs to bind a socket to a file for communicating with the universal profiling host agent." +
+                     "This configuration option can be used to change the location. " +
+                     "Note that the total path name (including the socket) must not exceed 100 characters due to OS restrictions.\n" +
+                     "If unset, the value of the `java.io.tmpdir` system property will be used.")
+        .build();
+
+    public boolean isEnabled() {
+        return enabled.get();
+    }
+
+    public long getBufferSize() {
+        return bufferSize.get();
+    }
+
+    public String getSocketDir() {
+        String dir = socketDir.get();
+        return dir == null || dir.isEmpty() ? System.getProperty("java.io.tmpdir") : dir;
+    }
+
+}
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/ElasticApmTracer.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/ElasticApmTracer.java
index 711601c319..abc6ec489f 100644
--- a/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/ElasticApmTracer.java
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/ElasticApmTracer.java
@@ -71,6 +71,7 @@
 import co.elastic.apm.agent.tracer.dispatch.HeaderGetter;
 import co.elastic.apm.agent.tracer.reference.ReferenceCounted;
 import co.elastic.apm.agent.tracer.reference.ReferenceCountedMap;
+import co.elastic.apm.agent.universalprofiling.UniversalProfilingIntegration;
 import co.elastic.apm.agent.util.DependencyInjectingServiceLoader;
 import co.elastic.apm.agent.util.ExecutorUtils;
 import com.dslplatform.json.JsonWriter;
@@ -143,6 +144,8 @@ protected ActiveStack initialValue() {
     private final SpanConfiguration spanConfiguration;
     private final List<ActivationListener> activationListeners;
     private final MetricRegistry metricRegistry;
+
+    private final UniversalProfilingIntegration profilingIntegration;
     private final ScheduledThreadPoolExecutor sharedPool;
     private final int approximateContextSize;
     private Sampler sampler;
@@ -261,6 +264,7 @@ public void onChange(ConfigurationOption<?> configurationOption, Double oldValue
         // sets the assertionsEnabled flag to true if indeed enabled
         //noinspection AssertWithSideEffects
         assert assertionsEnabled = true;
+        profilingIntegration = new UniversalProfilingIntegration();
     }
 
     @Override
@@ -341,6 +345,7 @@ private void afterTransactionStart(@Nullable ClassLoader initiatingClassLoader,
         if (serviceInfo != null) {
             transaction.getTraceContext().setServiceInfo(serviceInfo.getServiceName(), serviceInfo.getServiceVersion());
         }
+        profilingIntegration.afterTransactionStart(transaction);
     }
 
     public Transaction noopTransaction() {
@@ -525,8 +530,9 @@ public void endTransaction(Transaction transaction) {
         if (!transaction.isNoop() &&
             (transaction.isSampled() || apmServerClient.supportsKeepingUnsampledTransaction())) {
             // we do report non-sampled transactions (without the context)
-            reporter.report(transaction);
+            profilingIntegration.correlateAndReport(transaction);
         } else {
+            profilingIntegration.drop(transaction);
             transaction.decrementReferences();
         }
     }
@@ -633,6 +639,7 @@ public synchronized void stop() {
             logger.debug("Tracer stop stack trace: ", new Throwable("Expected - for debugging purposes"));
         }
 
+        profilingIntegration.stop();
         try {
             configurationRegistry.close();
             reporter.close();
@@ -738,6 +745,7 @@ private synchronized void startSync() {
         }
         apmServerClient.start();
         reporter.start();
+        profilingIntegration.start(this);
         for (LifecycleListener lifecycleListener : lifecycleListeners) {
             try {
                 lifecycleListener.start(this);
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/metadata/SystemInfo.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/metadata/SystemInfo.java
index 68624213ef..cb2c0f814a 100644
--- a/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/metadata/SystemInfo.java
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/metadata/SystemInfo.java
@@ -138,7 +138,7 @@ public static SystemInfo create(final @Nullable String configuredHostname, final
         return systemInfo.findContainerDetails();
     }
 
-    static boolean isWindows(String osName) {
+    public static boolean isWindows(String osName) {
         return osName.startsWith("Windows");
     }
 
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/Id.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/Id.java
index 27aa8289b7..5736124dce 100644
--- a/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/Id.java
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/Id.java
@@ -89,6 +89,10 @@ public int toBytes(byte[] bytes, int offset) {
         return offset + data.length;
     }
 
+    public void writeToBuffer(ByteBuffer buffer) {
+        buffer.put(data);
+    }
+
     public void fromLongs(long... values) {
         if (values.length * Long.BYTES != data.length) {
             throw new IllegalArgumentException("Invalid number of long values");
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/ProfilerSharedMemoryWriter.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/ProfilerSharedMemoryWriter.java
new file mode 100644
index 0000000000..259fb7687a
--- /dev/null
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/ProfilerSharedMemoryWriter.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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 co.elastic.apm.agent.universalprofiling;
+
+import co.elastic.apm.agent.impl.transaction.AbstractSpan;
+import co.elastic.apm.agent.impl.transaction.Transaction;
+import co.elastic.apm.agent.sdk.logging.Logger;
+import co.elastic.apm.agent.sdk.logging.LoggerFactory;
+import co.elastic.otel.UniversalProfilingCorrelation;
+
+import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+
+public class ProfilerSharedMemoryWriter {
+
+    private static final Logger log = LoggerFactory.getLogger(ProfilerSharedMemoryWriter.class);
+
+    private static final int TLS_MINOR_VERSION_OFFSET = 0;
+    private static final int TLS_VALID_OFFSET = 2;
+    private static final int TLS_TRACE_PRESENT_OFFSET = 3;
+    private static final int TLS_TRACE_FLAGS_OFFSET = 4;
+    private static final int TLS_TRACE_ID_OFFSET = 5;
+    private static final int TLS_SPAN_ID_OFFSET = 21;
+    private static final int TLS_LOCAL_ROOT_SPAN_ID_OFFSET = 29;
+    static final int TLS_STORAGE_SIZE = 37;
+
+    private static volatile int writeForMemoryBarrier = 0;
+
+    static ByteBuffer generateProcessCorrelationStorage(String serviceName, @Nullable String environment, String socketFilePath) {
+        ByteBuffer buffer = ByteBuffer.allocateDirect(4096);
+        buffer.order(ByteOrder.nativeOrder());
+        buffer.position(0);
+
+        buffer.putChar((char) 1); // layout-minor-version
+        writeUtf8Str(buffer, serviceName);
+        writeUtf8Str(buffer, environment == null ? "" : environment);
+        writeUtf8Str(buffer, socketFilePath);
+        return buffer;
+    }
+
+    private static void writeUtf8Str(ByteBuffer buffer, String str) {
+        byte[] utf8 = str.getBytes(StandardCharsets.UTF_8);
+        buffer.putInt(utf8.length);
+        buffer.put(utf8);
+    }
+
+    /**
+     * This method ensures that all writes which happened prior to this method call are not moved
+     * after the method call due to reordering.
+     *
+     * <p>This is realized based on the Java Memory Model guarantess for volatile variables. Relevant
+     * resources:
+     *
+     * <ul>
+     *   <li><a
+     *       href="https://stackoverflow.com/questions/17108541/happens-before-relationships-with-volatile-fields-and-synchronized-blocks-in-jav">StackOverflow
+     *       topic</a>
+     *   <li><a href="https://gee.cs.oswego.edu/dl/jmm/cookbook.html">JSR Compiler Cookbook</a>
+     * </ul>
+     */
+    private static void memoryStoreStoreBarrier() {
+        writeForMemoryBarrier = 42;
+    }
+
+    static void updateThreadCorrelationStorage(@Nullable AbstractSpan<?> newSpan) {
+        try {
+            ByteBuffer tls = UniversalProfilingCorrelation.getCurrentThreadStorage(true, TLS_STORAGE_SIZE);
+            // tls might be null if unsupported or something went wrong on initialization
+            if (tls != null) {
+                // the valid flag is used to signal the host-agent that it is reading incomplete data
+                tls.put(TLS_VALID_OFFSET, (byte) 0);
+                memoryStoreStoreBarrier();
+                tls.putChar(TLS_MINOR_VERSION_OFFSET, (char) 1);
+
+                if (newSpan != null) {
+                    Transaction tx = newSpan.getParentTransaction();
+                    tls.put(TLS_TRACE_PRESENT_OFFSET, (byte) 1);
+                    tls.put(TLS_TRACE_FLAGS_OFFSET, newSpan.getTraceContext().getFlags());
+                    tls.position(TLS_TRACE_ID_OFFSET);
+                    newSpan.getTraceContext().getTraceId().writeToBuffer(tls);
+                    tls.position(TLS_SPAN_ID_OFFSET);
+                    newSpan.getTraceContext().getId().writeToBuffer(tls);
+                    tls.position(TLS_LOCAL_ROOT_SPAN_ID_OFFSET);
+                    tx.getTraceContext().getId().writeToBuffer(tls);
+                } else {
+                    tls.put(TLS_TRACE_PRESENT_OFFSET, (byte) 0);
+                }
+                memoryStoreStoreBarrier();
+                tls.put(TLS_VALID_OFFSET, (byte) 1);
+            }
+        } catch (Exception e) {
+            log.error("Failed to write profiling correlation tls", e);
+        }
+    }
+}
+
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/UniversalProfilingIntegration.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/UniversalProfilingIntegration.java
new file mode 100644
index 0000000000..251aeaf6df
--- /dev/null
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/UniversalProfilingIntegration.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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 co.elastic.apm.agent.universalprofiling;
+
+import co.elastic.apm.agent.configuration.CoreConfiguration;
+import co.elastic.apm.agent.configuration.UniversalProfilingConfiguration;
+import co.elastic.apm.agent.impl.ActivationListener;
+import co.elastic.apm.agent.impl.ElasticApmTracer;
+import co.elastic.apm.agent.impl.metadata.SystemInfo;
+import co.elastic.apm.agent.impl.transaction.AbstractSpan;
+import co.elastic.apm.agent.impl.transaction.ElasticContext;
+import co.elastic.apm.agent.impl.transaction.Transaction;
+import co.elastic.apm.agent.sdk.logging.Logger;
+import co.elastic.apm.agent.sdk.logging.LoggerFactory;
+import co.elastic.otel.JvmtiAccess;
+import co.elastic.otel.UniversalProfilingCorrelation;
+
+import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+
+public class UniversalProfilingIntegration {
+
+    private static final Logger log = LoggerFactory.getLogger(UniversalProfilingIntegration.class);
+
+    private volatile ElasticApmTracer tracer;
+
+    // Visible for testing
+    volatile boolean isActive = false;
+
+    private ActivationListener activationListener = new ActivationListener() {
+
+        @Override
+        public void beforeActivate(AbstractSpan<?> span) {
+            ProfilerSharedMemoryWriter.updateThreadCorrelationStorage(span);
+        }
+
+        @Override
+        public void afterDeactivate(@Nullable AbstractSpan<?> deactivatedSpan) {
+            ElasticContext<?> currentContext = tracer.currentContext();
+            ProfilerSharedMemoryWriter.updateThreadCorrelationStorage(currentContext.getSpan());
+        }
+    };
+
+    public void start(ElasticApmTracer tracer) {
+        this.tracer = tracer;
+        UniversalProfilingConfiguration config = tracer.getConfig(UniversalProfilingConfiguration.class);
+        if (!config.isEnabled()) {
+            return;
+        }
+        if (SystemInfo.isWindows(System.getProperty("os.name"))) {
+            log.warn("Universal profiling integration is not supported on Windows");
+            return;
+        }
+        try {
+            log.debug("Starting universal profiling correlation");
+
+            CoreConfiguration coreConfig = tracer.getConfig(CoreConfiguration.class);
+            ByteBuffer processCorrelationStorage = ProfilerSharedMemoryWriter.generateProcessCorrelationStorage(
+                coreConfig.getServiceName(), coreConfig.getEnvironment(), "");
+            UniversalProfilingCorrelation.setProcessStorage(processCorrelationStorage);
+
+            isActive = true;
+            tracer.registerSpanListener(activationListener);
+        } catch (Exception e) {
+            log.error("Failed to start universal profiling integration", e);
+        }
+    }
+
+    public void stop() {
+        try {
+            if (isActive) {
+                JvmtiAccess.destroy();
+            }
+        } catch (Exception e) {
+            log.error("Failed to stop universal profiling integration", e);
+        }
+    }
+
+    public void afterTransactionStart(Transaction startedTransaction) {
+        //TODO: store the transaction in a map for correlating with profiling data
+    }
+
+    /**
+     * This method will get invoked for ended transactions which shall be reported.
+     * This method is responsible for eventually reporting those transactions.
+     * <p>
+     * The tracer calls this method instead of reporting directly in order to allow
+     * the correlation to delay the transaction reporting until all correlation
+     * data has been received from the universal profiling host agent.
+     *
+     * @param endedTransaction the transaction to be reported
+     */
+    public void correlateAndReport(Transaction endedTransaction) {
+        //TODO: perform correlation and report after buffering for a certain delay
+        tracer.getReporter().report(endedTransaction);
+    }
+
+    public void drop(Transaction endedTransaction) {
+        //TODO: remove dropped transactions from correlation storage without reporting
+    }
+}
diff --git a/apm-agent-core/src/main/resources/META-INF/services/org.stagemonitor.configuration.ConfigurationOptionProvider b/apm-agent-core/src/main/resources/META-INF/services/org.stagemonitor.configuration.ConfigurationOptionProvider
index 389c99067e..878419acfb 100644
--- a/apm-agent-core/src/main/resources/META-INF/services/org.stagemonitor.configuration.ConfigurationOptionProvider
+++ b/apm-agent-core/src/main/resources/META-INF/services/org.stagemonitor.configuration.ConfigurationOptionProvider
@@ -7,3 +7,4 @@ co.elastic.apm.agent.impl.circuitbreaker.CircuitBreakerConfiguration
 co.elastic.apm.agent.configuration.MetricsConfiguration
 co.elastic.apm.agent.configuration.ServerlessConfiguration
 co.elastic.apm.agent.configuration.SpanConfiguration
+co.elastic.apm.agent.configuration.UniversalProfilingConfiguration
diff --git a/apm-agent-core/src/test/java/co/elastic/apm/agent/universalprofiling/UniversalProfilingIntegrationTest.java b/apm-agent-core/src/test/java/co/elastic/apm/agent/universalprofiling/UniversalProfilingIntegrationTest.java
new file mode 100644
index 0000000000..867efae359
--- /dev/null
+++ b/apm-agent-core/src/test/java/co/elastic/apm/agent/universalprofiling/UniversalProfilingIntegrationTest.java
@@ -0,0 +1,230 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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 co.elastic.apm.agent.universalprofiling;
+
+import co.elastic.apm.agent.MockReporter;
+import co.elastic.apm.agent.MockTracer;
+import co.elastic.apm.agent.configuration.CoreConfiguration;
+import co.elastic.apm.agent.configuration.SpyConfiguration;
+import co.elastic.apm.agent.configuration.UniversalProfilingConfiguration;
+import co.elastic.apm.agent.impl.ElasticApmTracer;
+import co.elastic.apm.agent.impl.Tracer;
+import co.elastic.apm.agent.impl.transaction.AbstractSpan;
+import co.elastic.apm.agent.impl.transaction.Id;
+import co.elastic.apm.agent.impl.transaction.Span;
+import co.elastic.apm.agent.impl.transaction.Transaction;
+import co.elastic.apm.agent.objectpool.TestObjectPoolFactory;
+import co.elastic.otel.JvmtiAccessImpl;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.stagemonitor.configuration.ConfigurationRegistry;
+
+import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.util.function.Consumer;
+
+import static co.elastic.apm.agent.testutils.assertions.Assertions.assertThat;
+import static co.elastic.apm.agent.universalprofiling.ProfilerSharedMemoryWriter.TLS_STORAGE_SIZE;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+public class UniversalProfilingIntegrationTest {
+
+    private Tracer tracer;
+    private MockReporter reporter;
+
+    private TestObjectPoolFactory poolFactory;
+
+    void setupTracer(Consumer<ConfigurationRegistry> configCustomizer) {
+        ConfigurationRegistry config = SpyConfiguration.createSpyConfig();
+        configCustomizer.accept(config);
+        MockTracer.MockInstrumentationSetup mockInstrumentationSetup = MockTracer.createMockInstrumentationSetup(config);
+
+        tracer = mockInstrumentationSetup.getTracer();
+        reporter = mockInstrumentationSetup.getReporter();
+        poolFactory = mockInstrumentationSetup.getObjectPoolFactory();
+    }
+
+    @AfterEach
+    public void cleanupTracer() {
+        if (tracer != null) {
+            reporter.assertRecycledAfterDecrementingReferences();
+            poolFactory.checkAllPooledObjectsHaveBeenRecycled();
+            tracer.stop();
+            tracer = null;
+        }
+    }
+
+    @EnabledOnOs(OS.WINDOWS)
+    @Test
+    public void ensureDisabledOnWindows() {
+        ConfigurationRegistry configRegistry = SpyConfiguration.createSpyConfig();
+        UniversalProfilingConfiguration config = configRegistry.getConfig(UniversalProfilingConfiguration.class);
+
+        doReturn(true).when(config).isEnabled();
+
+        UniversalProfilingIntegration universalProfilingIntegration = new UniversalProfilingIntegration();
+        ElasticApmTracer mockTracer = MockTracer.create(configRegistry);
+        universalProfilingIntegration.start(mockTracer);
+
+        verify(mockTracer, never()).registerSpanListener(any());
+        assertThat(universalProfilingIntegration.isActive).isFalse();
+    }
+
+    @Test
+    public void ensureDisabledByDefault() {
+        ConfigurationRegistry configRegistry = SpyConfiguration.createSpyConfig();
+        UniversalProfilingIntegration universalProfilingIntegration = new UniversalProfilingIntegration();
+        ElasticApmTracer mockTracer = MockTracer.create(configRegistry);
+        universalProfilingIntegration.start(mockTracer);
+
+        verify(mockTracer, never()).registerSpanListener(any());
+        assertThat(universalProfilingIntegration.isActive).isFalse();
+    }
+
+    @Nested
+    @DisabledOnOs(OS.WINDOWS)
+    class SharedMemory {
+
+        @Test
+        public void testNestedActivations() {
+            setupTracer(conf -> doReturn(true)
+                .when(conf.getConfig(UniversalProfilingConfiguration.class)).isEnabled());
+
+            Transaction first = tracer.startRootTransaction(null);
+            Transaction second = tracer.startRootTransaction(null);
+            Span third = second.createSpan();
+
+            checkTlsIs(null);
+
+            first.activate();
+            {
+                checkTlsIs(first);
+
+                //nested activation
+                first.activate();
+                {
+                    checkTlsIs(first);
+                }
+                first.deactivate();
+                second.activate();
+                {
+                    checkTlsIs(second);
+                    third.activate();
+                    {
+                        checkTlsIs(third);
+
+                        first.activate();
+                        {
+                            checkTlsIs(first);
+                        }
+                        first.deactivate();
+                    }
+                    third.deactivate();
+                    checkTlsIs(second);
+                }
+                second.deactivate();
+                checkTlsIs(first);
+            }
+            first.deactivate();
+            checkTlsIs(null);
+
+            first.end();
+            second.end();
+            third.end();
+            assertThat(reporter.getTransactions()).containsExactly(first, second);
+            assertThat(reporter.getSpans()).containsExactly(third);
+        }
+
+
+        @ParameterizedTest
+        @ValueSource(strings = {"my nameßspace", ""})
+        public void testProcessStoragePopulated(String environment) {
+            setupTracer(conf -> {
+                doReturn(true).when(conf.getConfig(UniversalProfilingConfiguration.class)).isEnabled();
+                CoreConfiguration core = conf.getConfig(CoreConfiguration.class);
+                doReturn("service Ä 1").when(core).getServiceName();
+                doReturn(environment).when(core).getEnvironment();
+            });
+            ByteBuffer buffer = JvmtiAccessImpl.createProcessProfilingCorrelationBufferAlias(1000);
+            buffer.order(ByteOrder.nativeOrder());
+
+            assertThat(buffer.getChar()).describedAs("layout-minor-version").isEqualTo((char) 1);
+            assertThat(readUtf8Str(buffer)).isEqualTo("service Ä 1");
+            assertThat(readUtf8Str(buffer)).isEqualTo(environment);
+            assertThat(readUtf8Str(buffer)).describedAs("socket-path").isEqualTo("");
+        }
+
+        private String readUtf8Str(ByteBuffer buffer) {
+            int serviceNameLen = buffer.getInt();
+            byte[] serviceUtf8 = new byte[serviceNameLen];
+            buffer.get(serviceUtf8);
+            return new String(serviceUtf8, StandardCharsets.UTF_8);
+        }
+    }
+
+    static void checkTlsIs(@Nullable AbstractSpan<?> span) {
+        ByteBuffer tls = JvmtiAccessImpl.createThreadProfilingCorrelationBufferAlias(TLS_STORAGE_SIZE);
+        if (tls != null) {
+            tls.order(ByteOrder.nativeOrder());
+            Assertions.assertThat(tls.getChar(0)).describedAs("layout-minor-version").isEqualTo((char) 1);
+            Assertions.assertThat(tls.get(2)).describedAs("valid byte").isEqualTo((byte) 1);
+        }
+
+        if (span != null) {
+            assertThat(tls).isNotNull();
+            Assertions.assertThat(tls.get(3)).describedAs("trace-present-flag").isEqualTo((byte) 1);
+            Assertions.assertThat(tls.get(4)).describedAs("trace-flags").isEqualTo(span.getTraceContext().getFlags());
+
+            byte[] traceId = new byte[16];
+            byte[] spanId = new byte[8];
+            byte[] localRootSpanId = new byte[8];
+            tls.position(5);
+            tls.get(traceId);
+            assertThat(traceId).containsExactly(idToBytes(span.getTraceContext().getTraceId()));
+            tls.position(21);
+            tls.get(spanId);
+            assertThat(spanId).containsExactly(idToBytes(span.getTraceContext().getId()));
+            tls.position(29);
+            tls.get(localRootSpanId);
+            assertThat(localRootSpanId).containsExactly(idToBytes(span.getParentTransaction().getTraceContext().getId()));
+        } else if (tls != null) {
+            Assertions.assertThat(tls.get(3)).describedAs("trace-present-flag").isEqualTo((byte) 0);
+        }
+    }
+
+    private static byte[] idToBytes(Id id) {
+        byte[] buff = new byte[32];
+        int len = id.toBytes(buff, 0);
+        byte[] result = new byte[len];
+        System.arraycopy(buff, 0, result, 0, len);
+        return result;
+    }
+}