diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 0cdeb84dbe..4523676a8c 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -31,6 +31,10 @@ Use subheadings with the "=====" level for adding notes for unreleased changes:
 
 === Unreleased
 
+[float]
+===== Features
+* Added support for correlating APM data with elastic universal profiling data - {pull}3615[#3615], {pull}3602[#3602], {pull}3607[#3607], {pull}3598[#3598]
+
 [float]
 ===== Bug fixes
 * Fixed edge case where inferred spans could cause cycles in the trace parent-child relationships, subsequently resulting in the UI crashing - {pull}3588[#3588]
diff --git a/apm-agent-core/pom.xml b/apm-agent-core/pom.xml
index 5f35f7912d..b542555929 100644
--- a/apm-agent-core/pom.xml
+++ b/apm-agent-core/pom.xml
@@ -103,7 +103,7 @@
         <dependency>
             <groupId>co.elastic.otel</groupId>
             <artifactId>jvmti-access</artifactId>
-            <version>0.3.0</version>
+            <version>0.3.1</version>
         </dependency>
         <!--
         We can't use caffeine due to requiring Java 7.
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
index 98ca6b0cb5..5fd370495f 100644
--- 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
@@ -29,26 +29,26 @@ public class UniversalProfilingConfiguration extends ConfigurationOptionProvider
 
     private final ConfigurationOption<Boolean> enabled = ConfigurationOption.booleanOption()
         .key("universal_profiling_integration_enabled")
-        .tags("added[1.50.0]", "internal")
+        .tags("added[1.50.0]")
         .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()
+    private final ConfigurationOption<Integer> bufferSize = ConfigurationOption.integerOption()
         .key("universal_profiling_integration_buffer_size")
-        .addValidator(isInRange(64L, Long.MAX_VALUE))
-        .tags("added[1.50.0]", "internal")
+        .addValidator(isInRange(64, Integer.MAX_VALUE))
+        .tags("added[1.50.0]")
         .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);
+        .buildWithDefault(4096);
 
     private final ConfigurationOption<String> socketDir = ConfigurationOption.stringOption()
         .key("universal_profiling_integration_socket_dir")
-        .tags("added[1.50.0]", "internal")
+        .tags("added[1.50.0]")
         .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. " +
@@ -60,7 +60,7 @@ public boolean isEnabled() {
         return enabled.get();
     }
 
-    public long getBufferSize() {
+    public int getBufferSize() {
         return bufferSize.get();
     }
 
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 025de06219..d50ba17470 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
@@ -673,6 +673,10 @@ public synchronized void stop() {
     public Reporter getReporter() {
         return reporter;
     }
+    
+    public UniversalProfilingIntegration getProfilingIntegration() {
+        return profilingIntegration;
+    }
 
     public Sampler getSampler() {
         return sampler;
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/report/serialize/Base64SerializationUtils.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/report/serialize/Base64SerializationUtils.java
index aea9fde5bc..20bcb088bc 100644
--- a/apm-agent-core/src/main/java/co/elastic/apm/agent/report/serialize/Base64SerializationUtils.java
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/report/serialize/Base64SerializationUtils.java
@@ -1,3 +1,21 @@
+/*
+ * 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.report.serialize;
 
 import com.dslplatform.json.JsonWriter;
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/Clock.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/Clock.java
new file mode 100644
index 0000000000..11ea268016
--- /dev/null
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/Clock.java
@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+public interface Clock {
+
+    Clock SYSTEM_NANOTIME = new Clock() {
+        @Override
+        public long getNanos() {
+            return System.nanoTime();
+        }
+    };
+
+    long getNanos();
+}
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/MoveableEvent.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/MoveableEvent.java
new file mode 100644
index 0000000000..9fb79f7396
--- /dev/null
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/MoveableEvent.java
@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+public interface MoveableEvent<SELF extends MoveableEvent<?>> {
+
+    /**
+     * Moves the content from this event into the provided other event. This event should be in a
+     * resetted state after the call.
+     */
+    void moveInto(SELF other);
+
+    void clear();
+}
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/PeekingPoller.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/PeekingPoller.java
new file mode 100644
index 0000000000..772c24071a
--- /dev/null
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/PeekingPoller.java
@@ -0,0 +1,90 @@
+/*
+ * 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 com.lmax.disruptor.EventFactory;
+import com.lmax.disruptor.EventPoller;
+
+import java.util.function.Supplier;
+
+/**
+ * Wrapper around {@link EventPoller} which allows to "peek" elements. The provided event handling
+ * callback can decide to not handle an event. In that case, the event will be provided again as
+ * first element on the next call to {@link #poll(Handler)}.
+ */
+public class PeekingPoller<Event extends MoveableEvent<Event>> {
+
+    public interface Handler<Event extends MoveableEvent<Event>> {
+
+        /**
+         * Handles an event fetched from the ring buffer.
+         *
+         * @return true, if the event was handled and shall be removed. False if the event was not
+         * handled, no further invocations of handleEvent are desired and the same event shall be
+         * provided for the next {@link PeekingPoller#poll(Handler)} call.
+         */
+        boolean handleEvent(Event e);
+    }
+
+    private final EventPoller<Event> poller;
+    private final Event peekedEvent;
+    private boolean peekedEventPopulated;
+
+    Handler<? super Event> currentHandler;
+    private final EventPoller.Handler<Event> subHandler = new EventPoller.Handler<Event>() {
+        @Override
+        public boolean onEvent(Event event, long sequence, boolean endOfBatch) throws Exception {
+            return handleEvent(event, sequence, endOfBatch);
+        }
+    };
+
+    public PeekingPoller(EventPoller<Event> wrappedPoller, EventFactory<Event> emptyEventFactory) {
+        this.poller = wrappedPoller;
+        peekedEvent = emptyEventFactory.newInstance();
+        peekedEventPopulated = false;
+    }
+
+    public synchronized void poll(Handler<? super Event> handler) throws Exception {
+        if (peekedEventPopulated) {
+            boolean handled = handler.handleEvent(peekedEvent);
+            if (!handled) {
+                return;
+            }
+            peekedEvent.clear();
+            peekedEventPopulated = false;
+        }
+        currentHandler = handler;
+        try {
+            poller.poll(subHandler);
+        } finally {
+            currentHandler = null;
+        }
+    }
+
+    private boolean handleEvent(Event event, long sequence, boolean endOfBatch) {
+        boolean handled = currentHandler.handleEvent(event);
+        if (handled) {
+            return true;
+        } else {
+            peekedEventPopulated = true;
+            event.moveInto(peekedEvent);
+            return false;
+        }
+    }
+}
diff --git a/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/SpanProfilingSamplesCorrelator.java b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/SpanProfilingSamplesCorrelator.java
new file mode 100644
index 0000000000..90d6b36981
--- /dev/null
+++ b/apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/SpanProfilingSamplesCorrelator.java
@@ -0,0 +1,218 @@
+/*
+ * 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.Id;
+import co.elastic.apm.agent.impl.transaction.Transaction;
+import co.elastic.apm.agent.report.Reporter;
+import co.elastic.apm.agent.sdk.logging.Logger;
+import co.elastic.apm.agent.sdk.logging.LoggerFactory;
+import com.lmax.disruptor.EventFactory;
+import com.lmax.disruptor.EventPoller;
+import com.lmax.disruptor.EventTranslatorTwoArg;
+import com.lmax.disruptor.RingBuffer;
+import com.lmax.disruptor.YieldingWaitStrategy;
+import org.HdrHistogram.WriterReaderPhaser;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+public class SpanProfilingSamplesCorrelator {
+
+    private static final Logger logger = LoggerFactory.getLogger(SpanProfilingSamplesCorrelator.class);
+
+    /**
+     * Holds the currently active transactions by their span-ID.
+     * Note that theoretically there could be a collision by having two transactions with different trace-IDs but the same span-ID.
+     * However, this is highly unlikely (see <a href="https://en.wikipedia.org/wiki/Birthday_problem">birthday problem</a>) and even if
+     * it were to happen, the only consequences would be a potentially incorrect correlation for the two colliding transactions.
+     */
+    private final ConcurrentHashMap<Id, Transaction> transactionsById = new ConcurrentHashMap<>();
+
+    private final Reporter reporter;
+
+    // Clock to use, can be swapped out for testing
+    Clock nanoClock = Clock.SYSTEM_NANOTIME;
+
+    // Visible for testing
+    final RingBuffer<BufferedTransaction> delayedSpans;
+    private final PeekingPoller<BufferedTransaction> delayedSpansPoller;
+
+    private volatile long spanBufferDurationNanos;
+    private volatile boolean shuttingDown = false;
+
+    private final WriterReaderPhaser shutdownPhaser = new WriterReaderPhaser();
+
+    public SpanProfilingSamplesCorrelator(
+        int bufferCapacity,
+        long initialSpanDelayNanos,
+        Reporter reporter) {
+        this.spanBufferDurationNanos = initialSpanDelayNanos;
+        this.reporter = reporter;
+
+        bufferCapacity = nextPowerOf2(bufferCapacity);
+        // We use a wait strategy which doesn't involve signaling via condition variables
+        // because we never block anyway (we use polling)
+        EventFactory<BufferedTransaction> eventFactory = new EventFactory<BufferedTransaction>() {
+            @Override
+            public BufferedTransaction newInstance() {
+                return new BufferedTransaction();
+            }
+        };
+        delayedSpans = RingBuffer.createMultiProducer(eventFactory, bufferCapacity, new YieldingWaitStrategy());
+        EventPoller<BufferedTransaction> nonPeekingPoller = delayedSpans.newPoller();
+        delayedSpans.addGatingSequences(nonPeekingPoller.getSequence());
+
+        delayedSpansPoller = new PeekingPoller<>(nonPeekingPoller, eventFactory);
+    }
+
+    public void setSpanBufferDurationNanos(long nanos) {
+        if (nanos < 0) {
+            throw new IllegalArgumentException("nanos must be positive but was " + nanos);
+        }
+        spanBufferDurationNanos = nanos;
+    }
+
+    public void onTransactionStart(Transaction transaction) {
+        if (transaction.isSampled()) {
+            transactionsById.put(transaction.getTraceContext().getId(), transaction);
+        }
+    }
+
+    public void stopCorrelating(Transaction transaction) {
+        transactionsById.remove(transaction.getTraceContext().getId());
+    }
+
+    public void reportOrBufferTransaction(Transaction transaction) {
+        if (!transactionsById.containsKey(transaction.getTraceContext().getId())) {
+            // transaction is not being correlated, e.g. because it was not sampled
+            // therefore no need to buffer it
+            reporter.report(transaction);
+            return;
+        }
+
+        long criticalPhaseVal = shutdownPhaser.writerCriticalSectionEnter();
+        try {
+            if (spanBufferDurationNanos == 0 || shuttingDown) {
+                reporter.report(transaction);
+                return;
+            }
+
+            boolean couldPublish =
+                delayedSpans.tryPublishEvent(BufferedTransaction.TRANSLATOR,
+                    transaction,
+                    nanoClock.getNanos());
+
+            if (!couldPublish) {
+                logger.warn("The following transaction could not be delayed for correlation due to a full buffer, it will be sent immediately, {0}",
+                    transaction);
+                reporter.report(transaction);
+            }
+        } finally {
+            shutdownPhaser.writerCriticalSectionExit(criticalPhaseVal);
+        }
+    }
+
+    public synchronized void correlate(
+        Id traceId, Id transactionId, Id stackTraceId, int count) {
+        Transaction tx = transactionsById.get(transactionId);
+        if (tx != null) {
+            // this branch should be true practically always unless there was a collision in transactionsById
+            // nonetheless for the unlikely case that it happens, we at least prevent wrongly adding data to another transaction
+            if (tx.getTraceContext().getTraceId().equals(traceId)) {
+                for (int i = 0; i < count; i++) {
+                    tx.addProfilerCorrelationStackTrace(stackTraceId);
+                }
+            }
+        }
+    }
+
+    private final PeekingPoller.Handler<BufferedTransaction> BUFFERED_TRANSACTION_HANDLER = new PeekingPoller.Handler<BufferedTransaction>() {
+        @Override
+        public boolean handleEvent(BufferedTransaction bufferedSpan) {
+            long elapsed = nanoClock.getNanos() - bufferedSpan.endNanoTimestamp;
+            if (elapsed >= spanBufferDurationNanos || shuttingDown) {
+                stopCorrelating(bufferedSpan.transaction);
+                reporter.report(bufferedSpan.transaction);
+                bufferedSpan.clear();
+                return true;
+            }
+            return false; // span is not yet ready to be sent
+        }
+    };
+
+    public synchronized void flushPendingBufferedSpans() {
+        try {
+            delayedSpansPoller.poll(BUFFERED_TRANSACTION_HANDLER);
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+    public synchronized void shutdownAndFlushAll() {
+        shutdownPhaser.readerLock();
+        try {
+            shuttingDown = true; // This will cause new ended spans to not be buffered anymore
+
+            // avoid race condition: we wait until we are
+            // sure that no more spans will be added to the ringbuffer
+            shutdownPhaser.flipPhase();
+        } finally {
+            shutdownPhaser.readerUnlock();
+        }
+        // This will flush all pending spans because shuttingDown=true
+        flushPendingBufferedSpans();
+    }
+
+    private static class BufferedTransaction implements MoveableEvent<BufferedTransaction> {
+
+        Transaction transaction;
+        long endNanoTimestamp;
+
+        @Override
+        public void moveInto(BufferedTransaction other) {
+            other.transaction = transaction;
+            other.endNanoTimestamp = endNanoTimestamp;
+            clear();
+        }
+
+        @Override
+        public void clear() {
+            transaction = null;
+            endNanoTimestamp = -1;
+        }
+
+        public static final EventTranslatorTwoArg<BufferedTransaction, Transaction, Long> TRANSLATOR = new EventTranslatorTwoArg<BufferedTransaction, Transaction, Long>() {
+            @Override
+            public void translateTo(BufferedTransaction event, long sequence, Transaction transaction, Long timestamp) {
+                event.transaction = transaction;
+                event.endNanoTimestamp = timestamp;
+            }
+        };
+    }
+
+    private static int nextPowerOf2(int val) {
+        int result = 2;
+        while (result < val) {
+            result *= 2;
+        }
+        return result;
+    }
+}
+
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
index 33897cbfa6..accfe2b8d2 100644
--- 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
@@ -25,6 +25,7 @@
 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.Id;
 import co.elastic.apm.agent.impl.transaction.Transaction;
 import co.elastic.apm.agent.sdk.logging.Logger;
 import co.elastic.apm.agent.sdk.logging.LoggerFactory;
@@ -42,11 +43,9 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.time.Duration;
-import java.util.Base64;
 import java.util.Random;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
-import java.util.logging.Level;
 
 public class UniversalProfilingIntegration {
 
@@ -56,6 +55,8 @@ public class UniversalProfilingIntegration {
      */
     static final long POLL_FREQUENCY_MS = 20;
 
+    private static final long INITIAL_SPAN_DELAY_NANOS = 1_000_000_000L;
+
     private static final Logger log = LoggerFactory.getLogger(UniversalProfilingIntegration.class);
 
     private volatile ElasticApmTracer tracer;
@@ -66,8 +67,13 @@ public class UniversalProfilingIntegration {
     // Visible for testing
     String socketPath = null;
 
+    @Nullable
     private ScheduledExecutorService executor;
 
+    // Visible for testing
+    @Nullable
+    SpanProfilingSamplesCorrelator correlator;
+
     private ActivationListener activationListener = new ActivationListener() {
 
         @Override
@@ -102,6 +108,8 @@ public void start(ElasticApmTracer tracer) {
                 coreConfig.getServiceName(), coreConfig.getEnvironment(), socketPath);
             UniversalProfilingCorrelation.setProcessStorage(processCorrelationStorage);
 
+            correlator = new SpanProfilingSamplesCorrelator(config.getBufferSize(), INITIAL_SPAN_DELAY_NANOS, tracer.getReporter());
+
             executor = ExecutorUtils.createSingleThreadSchedulingDaemonPool("profiling-integration");
             executor.scheduleWithFixedDelay(new Runnable() {
                 @Override
@@ -125,8 +133,10 @@ public void run() {
         }
     }
 
-    private void periodicTimer() {
+    // Visible for testing
+    void periodicTimer() {
         consumeProfilerMessages();
+        correlator.flushPendingBufferedSpans();
     }
 
     public void stop() {
@@ -137,6 +147,8 @@ public void stop() {
                 executor = null;
             }
             if (isActive) {
+                consumeProfilerMessages();
+                correlator.shutdownAndFlushAll();
                 UniversalProfilingCorrelation.stopProfilerReturnChannel();
                 JvmtiAccess.destroy();
                 isActive = false;
@@ -147,7 +159,9 @@ public void stop() {
     }
 
     public void afterTransactionStart(Transaction startedTransaction) {
-        //TODO: store the transaction in a map for correlating with profiling data
+        if (correlator != null) {
+            correlator.onTransactionStart(startedTransaction);
+        }
     }
 
     /**
@@ -161,12 +175,17 @@ public void afterTransactionStart(Transaction startedTransaction) {
      * @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);
+        if (correlator != null) {
+            correlator.reportOrBufferTransaction(endedTransaction);
+        } else {
+            tracer.getReporter().report(endedTransaction);
+        }
     }
 
     public void drop(Transaction endedTransaction) {
-        //TODO: remove dropped transactions from correlation storage without reporting
+        if (correlator != null) {
+            correlator.stopCorrelating(endedTransaction);
+        }
     }
 
 
@@ -197,7 +216,7 @@ private String randomSocketFileName() {
         return name.toString();
     }
 
-    private void consumeProfilerMessages() {
+    private synchronized void consumeProfilerMessages() {
         try {
             while (true) {
                 try {
@@ -223,10 +242,22 @@ private void consumeProfilerMessages() {
     }
 
     private void handleMessage(ProfilerRegistrationMessage message) {
-        //TODO: handle message
+        //TODO: update the host.id in the reporter metadata
+        log.debug("Received profiler registration message with host.id={} and expected latency of {} millis",
+            message.getHostId(), message.getSamplesDelayMillis());
+        long delayMillis = message.getSamplesDelayMillis() + POLL_FREQUENCY_MS;
+        correlator.setSpanBufferDurationNanos(delayMillis * 1_000_000L);
     }
 
+    private final Id tempTraceId = Id.new128BitId();
+    private final Id tempSpanId = Id.new64BitId();
+    private final Id tempStackTraceId = Id.new128BitId();
     private void handleMessage(TraceCorrelationMessage message) {
-        //TODO: handle message
+        tempTraceId.fromBytes(message.getTraceId(), 0);
+        tempSpanId.fromBytes(message.getLocalRootSpanId(), 0);
+        tempStackTraceId.fromBytes(message.getStackTraceId(), 0);
+        log.trace("Received profiler correlation message with trace.id={} transaction.id={} stacktrace.id={} count={}",
+            tempTraceId, tempSpanId, tempStackTraceId, message.getSampleCount());
+        correlator.correlate(tempTraceId, tempSpanId, tempStackTraceId, message.getSampleCount());
     }
 }
diff --git a/apm-agent-core/src/test/java/co/elastic/apm/agent/report/serialize/Base64SerializationUtilTest.java b/apm-agent-core/src/test/java/co/elastic/apm/agent/report/serialize/Base64SerializationUtilTest.java
index 257d6fba96..b16f295085 100644
--- a/apm-agent-core/src/test/java/co/elastic/apm/agent/report/serialize/Base64SerializationUtilTest.java
+++ b/apm-agent-core/src/test/java/co/elastic/apm/agent/report/serialize/Base64SerializationUtilTest.java
@@ -1,3 +1,21 @@
+/*
+ * 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.report.serialize;
 
 import com.dslplatform.json.DslJson;
diff --git a/apm-agent-core/src/test/java/co/elastic/apm/agent/universalprofiling/PeekingPollerTest.java b/apm-agent-core/src/test/java/co/elastic/apm/agent/universalprofiling/PeekingPollerTest.java
new file mode 100644
index 0000000000..75855a7368
--- /dev/null
+++ b/apm-agent-core/src/test/java/co/elastic/apm/agent/universalprofiling/PeekingPollerTest.java
@@ -0,0 +1,130 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+
+import com.lmax.disruptor.EventPoller;
+import com.lmax.disruptor.RingBuffer;
+import com.lmax.disruptor.YieldingWaitStrategy;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.Test;
+
+public class PeekingPollerTest {
+
+    private static class DummyEvent implements MoveableEvent<DummyEvent> {
+
+        private int val;
+
+        public DummyEvent() {
+            this(-1);
+        }
+
+        public DummyEvent(int val) {
+            this.val = val;
+        }
+
+        @Override
+        public void moveInto(DummyEvent other) {
+            other.val = val;
+            clear();
+        }
+
+        @Override
+        public void clear() {
+            val = -1;
+        }
+    }
+
+    private static class RecordingHandler implements PeekingPoller.Handler<DummyEvent> {
+
+        List<Integer> invocations = new ArrayList<>();
+        Function<Integer, Boolean> resultProvider;
+
+        @Override
+        public boolean handleEvent(DummyEvent event) {
+            invocations.add(event.val);
+            return resultProvider.apply(event.val);
+        }
+
+        void reset() {
+            invocations.clear();
+        }
+    }
+
+    @Test
+    public void testPeekingFunction() throws Exception {
+        RingBuffer<DummyEvent> rb =
+            RingBuffer.createMultiProducer(DummyEvent::new, 4, new YieldingWaitStrategy());
+        EventPoller<DummyEvent> nonPeekingPoller = rb.newPoller();
+        rb.addGatingSequences(nonPeekingPoller.getSequence());
+
+        PeekingPoller<DummyEvent> poller =
+            new PeekingPoller<DummyEvent>(nonPeekingPoller, DummyEvent::new);
+
+        RecordingHandler handler = new RecordingHandler();
+        handler.resultProvider = val -> false;
+        poller.poll(handler);
+        assertThat(handler.invocations).isEmpty();
+
+        assertThat(rb.tryPublishEvent((ev, srq) -> ev.val = 1)).isTrue();
+        assertThat(rb.tryPublishEvent((ev, srq) -> ev.val = 2)).isTrue();
+        assertThat(rb.tryPublishEvent((ev, srq) -> ev.val = 3)).isTrue();
+        assertThat(rb.tryPublishEvent((ev, srq) -> ev.val = 4)).isTrue();
+        assertThat(rb.tryPublishEvent((ev, srq) -> ev.val = 5)).isFalse();
+
+        poller.poll(handler);
+        assertThat(handler.invocations).containsExactly(1);
+        poller.poll(handler);
+        assertThat(handler.invocations).containsExactly(1, 1);
+
+        // It should now be possible to add one more element to the buffer
+        assertThat(rb.tryPublishEvent((ev, srq) -> ev.val = 5)).isTrue();
+        assertThat(rb.tryPublishEvent((ev, srq) -> ev.val = 6)).isFalse();
+
+        poller.poll(handler);
+        assertThat(handler.invocations).containsExactly(1, 1, 1);
+
+        // now consume elements up to index 2
+        handler.reset();
+        handler.resultProvider = i -> i != 3;
+
+        poller.poll(handler);
+        assertThat(handler.invocations).containsExactly(1, 2, 3);
+
+        // It should now be possible to add two more element to the buffer
+        assertThat(rb.tryPublishEvent((ev, srq) -> ev.val = 6)).isTrue();
+        assertThat(rb.tryPublishEvent((ev, srq) -> ev.val = 7)).isTrue();
+        assertThat(rb.tryPublishEvent((ev, srq) -> ev.val = 8)).isFalse();
+
+        poller.poll(handler);
+        assertThat(handler.invocations).containsExactly(1, 2, 3, 3);
+
+        // drain remaining elements
+        handler.reset();
+        handler.resultProvider = i -> true;
+
+        poller.poll(handler);
+        assertThat(handler.invocations).containsExactly(3, 4, 5, 6, 7);
+    }
+}
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
index 81b64c67cb..d9ec6cfb36 100644
--- 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
@@ -24,7 +24,7 @@
 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.sampling.ConstantSampler;
 import co.elastic.apm.agent.impl.transaction.AbstractSpan;
 import co.elastic.apm.agent.impl.transaction.Id;
 import co.elastic.apm.agent.impl.transaction.Span;
@@ -50,10 +50,14 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.time.Duration;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicLong;
 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.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.never;
@@ -61,7 +65,7 @@
 
 public class UniversalProfilingIntegrationTest {
 
-    private Tracer tracer;
+    private ElasticApmTracer tracer;
     private MockReporter reporter;
 
     private TestObjectPoolFactory poolFactory;
@@ -79,7 +83,7 @@ void setupTracer() {
     void setupTracer(Consumer<ConfigurationRegistry> configCustomizer) {
         ConfigurationRegistry config = SpyConfiguration.createSpyConfig();
         configCustomizer.accept(config);
-        MockTracer.MockInstrumentationSetup mockInstrumentationSetup = MockTracer.createMockInstrumentationSetup(config);
+        MockTracer.MockInstrumentationSetup mockInstrumentationSetup = MockTracer.createMockInstrumentationSetup(config, false);
 
         tracer = mockInstrumentationSetup.getTracer();
         reporter = mockInstrumentationSetup.getReporter();
@@ -89,9 +93,11 @@ void setupTracer(Consumer<ConfigurationRegistry> configCustomizer) {
     @AfterEach
     public void cleanupTracer() {
         if (tracer != null) {
+            if (tracer.isRunning()) {
+                tracer.stop();
+            }
             reporter.assertRecycledAfterDecrementingReferences();
             poolFactory.checkAllPooledObjectsHaveBeenRecycled();
-            tracer.stop();
             tracer = null;
         }
     }
@@ -174,8 +180,6 @@ public void testNestedActivations() {
             first.end();
             second.end();
             third.end();
-            assertThat(reporter.getTransactions()).containsExactly(first, second);
-            assertThat(reporter.getSpans()).containsExactly(third);
         }
 
 
@@ -242,6 +246,137 @@ static void checkTlsIs(@Nullable AbstractSpan<?> span) {
     @Nested
     class SpanCorrelation {
 
+
+        @Test
+        void checkCorrelationFunctional() {
+            AtomicLong clockMs = new AtomicLong(0L);
+            setupTracer();
+            UniversalProfilingIntegration profilingIntegration = tracer.getProfilingIntegration();
+            profilingIntegration.correlator.nanoClock = () -> clockMs.get() * 1_000_000L;
+
+            sendProfilerRegistrationMsg(1, "hostid");
+
+            Transaction tx1 = tracer.startRootTransaction(null);
+            Transaction tx2 = tracer.startRootTransaction(null);
+
+            Id st1 = randomStackTraceId(1);
+            sendSampleMsg(tx1, st1, 1);
+
+            // Send some garbage which should not affect our processing
+            JvmtiAccessImpl.sendToProfilerReturnChannelSocket0(new byte[]{1, 2, 3});
+
+            Id st2 = randomStackTraceId(2);
+            sendSampleMsg(tx2, st2, 2);
+
+            // ensure that the messages are processed now
+            profilingIntegration.periodicTimer();
+
+            tx1.end();
+            tx2.end();
+
+            // ensure that spans are not sent, their delay has not yet elapsed
+            assertThat(reporter.getTransactions()).isEmpty();
+
+            Id st3 = randomStackTraceId(3);
+            sendSampleMsg(tx2, st2, 1);
+            sendSampleMsg(tx1, st3, 2);
+            sendSampleMsg(tx2, st3, 1);
+
+            clockMs.set(1L + UniversalProfilingIntegration.POLL_FREQUENCY_MS);
+            // now the background thread should consume those messages and flush the spans
+            await()
+                .atMost(Duration.ofSeconds(10))
+                .untilAsserted(
+                    () -> {
+
+                        assertThat(reporter.getTransactions())
+                            .hasSize(2)
+                            .containsExactlyInAnyOrder(tx1, tx2);
+
+                        assertThat(tx1.getProfilingCorrelationStackTraceIds())
+                            .containsExactlyInAnyOrder(st1, st3, st3);
+
+                        assertThat(tx2.getProfilingCorrelationStackTraceIds())
+                            .containsExactlyInAnyOrder(st2, st2, st2, st3);
+                    });
+        }
+
+        @Test
+        void unsampledTransactionsNotCorrelated() {
+            setupTracer();
+            UniversalProfilingIntegration profilingIntegration = tracer.getProfilingIntegration();
+            profilingIntegration.correlator.nanoClock = () -> 0L;
+            doReturn(true).when(tracer.getApmServerClient()).supportsKeepingUnsampledTransaction();
+
+            sendProfilerRegistrationMsg(1, "hostid");
+
+            Transaction tx = tracer.startRootTransaction(ConstantSampler.of(false), 0L, null);
+            assertThat(tx.isSampled()).isFalse();
+
+            // Still send a stacktrace to make sure it is actually ignored
+            sendSampleMsg(tx, randomStackTraceId(1), 1);
+
+            // ensure that the messages are processed now
+            profilingIntegration.periodicTimer();
+            tx.end();
+
+            assertThat(reporter.getTransactions()).containsExactly(tx);
+            assertThat(tx.getProfilingCorrelationStackTraceIds()).isEmpty();
+        }
+
+        @Test
+        void shutdownFlushesBufferedSpans() {
+            Id st1 = randomStackTraceId(1);
+
+            setupTracer();
+            UniversalProfilingIntegration profilingIntegration = tracer.getProfilingIntegration();
+            profilingIntegration.correlator.nanoClock = () -> 0L;
+
+            sendProfilerRegistrationMsg(1, "hostid");
+            profilingIntegration.periodicTimer();
+
+            Transaction tx = tracer.startRootTransaction(null);
+            tx.end();
+
+            profilingIntegration.periodicTimer();
+            assertThat(reporter.getTransactions()).isEmpty();
+
+            sendSampleMsg(tx, st1, 1);
+            tracer.stop();
+
+            assertThat(reporter.getTransactions()).containsExactly(tx);
+            assertThat(tx.getProfilingCorrelationStackTraceIds()).containsExactly(st1);
+        }
+
+        @Test
+        void bufferCapacityExceeded() {
+            setupTracer(conf -> {
+                UniversalProfilingConfiguration profConfig = conf.getConfig(UniversalProfilingConfiguration.class);
+                doReturn(true).when(profConfig).isEnabled();
+                doReturn(tempDir.toAbsolutePath().toString()).when(profConfig).getSocketDir();
+                doReturn(2).when(profConfig).getBufferSize();
+            });
+            UniversalProfilingIntegration profilingIntegration = tracer.getProfilingIntegration();
+            profilingIntegration.correlator.nanoClock = () -> 0L;
+
+
+            sendProfilerRegistrationMsg(1, "hostid");
+            profilingIntegration.periodicTimer();
+            Transaction tx1 = tracer.startRootTransaction(null);
+            tx1.end();
+            Transaction tx2 = tracer.startRootTransaction(null);
+            tx2.end();
+            //the actual buffer capacity is 2 + 1 because the peeked transaction is stored outside of the buffer
+            profilingIntegration.periodicTimer();
+            Transaction tx3 = tracer.startRootTransaction(null);
+            tx3.end();
+            // now the buffer should be full, transaction 4 should be sent immediately
+            Transaction tx4 = tracer.startRootTransaction(null);
+            tx4.end();
+
+            Assertions.assertThat(reporter.getTransactions()).containsExactly(tx4);
+        }
+
         @Test
         void badSocketPath() throws Exception {
             Path notADir = tempDir.resolve("not_a_dir");
@@ -287,8 +422,45 @@ void socketParentDirCreated() throws Exception {
             }
         }
 
-    }
 
+        private Id randomStackTraceId(int seed) {
+            byte[] id = new byte[16];
+            new Random(seed).nextBytes(id);
+            Id idObj = Id.new128BitId();
+            idObj.fromBytes(id, 0);
+            return idObj;
+        }
+
+        void sendSampleMsg(Transaction transaction, Id stackTraceId, int count) {
+            byte[] traceId = idToBytes(transaction.getTraceContext().getTraceId());
+            byte[] transactionId = idToBytes(transaction.getTraceContext().getId());
+
+            ByteBuffer message = ByteBuffer.allocate(46);
+            message.order(ByteOrder.nativeOrder());
+            message.putShort((short) 1); // message-type = correlation message
+            message.putShort((short) 1); // message-version
+            message.put(traceId);
+            message.put(transactionId);
+            message.put(idToBytes(stackTraceId));
+            message.putShort((short) count);
+
+            JvmtiAccessImpl.sendToProfilerReturnChannelSocket0(message.array());
+        }
+
+        void sendProfilerRegistrationMsg(int sampleDelayMillis, String hostId) {
+            byte[] hostIdUtf8 = hostId.getBytes(StandardCharsets.UTF_8);
+
+            ByteBuffer message = ByteBuffer.allocate(12 + hostIdUtf8.length);
+            message.order(ByteOrder.nativeOrder());
+            message.putShort((short) 2); // message-type = registration message
+            message.putShort((short) 1); // message-version
+            message.putInt(sampleDelayMillis);
+            message.putInt(hostIdUtf8.length);
+            message.put(hostIdUtf8);
+
+            JvmtiAccessImpl.sendToProfilerReturnChannelSocket0(message.array());
+        }
+    }
 
     private static byte[] idToBytes(Id id) {
         byte[] buff = new byte[32];
diff --git a/docs/configuration.asciidoc b/docs/configuration.asciidoc
index e493f21612..1b3bd263bd 100644
--- a/docs/configuration.asciidoc
+++ b/docs/configuration.asciidoc
@@ -179,6 +179,9 @@ Click on a key to get more information.
 ** <<config-agent-reporter-health-metrics>>
 ** <<config-agent-background-overhead-metrics>>
 * <<config-profiling>>
+** <<config-universal-profiling-integration-enabled>>
+** <<config-universal-profiling-integration-buffer-size>>
+** <<config-universal-profiling-integration-socket-dir>>
 ** <<config-profiling-inferred-spans-enabled>>
 ** <<config-profiling-inferred-spans-logging-enabled>>
 ** <<config-profiling-inferred-spans-sampling-interval>>
@@ -2673,6 +2676,77 @@ Enables metrics which capture the resource consumption of agent background tasks
 <titleabbrev>Profiling</titleabbrev>
 ++++
 
+// This file is auto generated. Please make your changes in *Configuration.java (for example CoreConfiguration.java) and execute ConfigurationExporter
+[float]
+[[config-universal-profiling-integration-enabled]]
+==== `universal_profiling_integration_enabled` (added[1.50.0])
+
+If enabled, the apm agent will correlate it's transaction with the profiling data from elastic universal profiling running on the same host.
+
+
+
+
+[options="header"]
+|============
+| Default                          | Type                | Dynamic
+| `false` | Boolean | false
+|============
+
+
+[options="header"]
+|============
+| Java System Properties      | Property file   | Environment
+| `elastic.apm.universal_profiling_integration_enabled` | `universal_profiling_integration_enabled` | `ELASTIC_APM_UNIVERSAL_PROFILING_INTEGRATION_ENABLED`
+|============
+
+// This file is auto generated. Please make your changes in *Configuration.java (for example CoreConfiguration.java) and execute ConfigurationExporter
+[float]
+[[config-universal-profiling-integration-buffer-size]]
+==== `universal_profiling_integration_buffer_size` (added[1.50.0])
+
+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.
+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.
+
+
+
+
+[options="header"]
+|============
+| Default                          | Type                | Dynamic
+| `4096` | Integer | false
+|============
+
+
+[options="header"]
+|============
+| Java System Properties      | Property file   | Environment
+| `elastic.apm.universal_profiling_integration_buffer_size` | `universal_profiling_integration_buffer_size` | `ELASTIC_APM_UNIVERSAL_PROFILING_INTEGRATION_BUFFER_SIZE`
+|============
+
+// This file is auto generated. Please make your changes in *Configuration.java (for example CoreConfiguration.java) and execute ConfigurationExporter
+[float]
+[[config-universal-profiling-integration-socket-dir]]
+==== `universal_profiling_integration_socket_dir` (added[1.50.0])
+
+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.
+If unset, the value of the `java.io.tmpdir` system property will be used.
+
+
+
+
+[options="header"]
+|============
+| Default                          | Type                | Dynamic
+| `<none>` | String | false
+|============
+
+
+[options="header"]
+|============
+| Java System Properties      | Property file   | Environment
+| `elastic.apm.universal_profiling_integration_socket_dir` | `universal_profiling_integration_socket_dir` | `ELASTIC_APM_UNIVERSAL_PROFILING_INTEGRATION_SOCKET_DIR`
+|============
+
 // This file is auto generated. Please make your changes in *Configuration.java (for example CoreConfiguration.java) and execute ConfigurationExporter
 [float]
 [[config-profiling-inferred-spans-enabled]]
@@ -4694,6 +4768,32 @@ Example: `5ms`.
 # Profiling                                #
 ############################################
 
+# If enabled, the apm agent will correlate it's transaction with the profiling data from elastic universal profiling running on the same host.
+#
+# This setting can not be changed at runtime. Changes require a restart of the application.
+# Type: Boolean
+# Default value: false
+#
+# universal_profiling_integration_enabled=false
+
+# 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.
+# 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.
+#
+# This setting can not be changed at runtime. Changes require a restart of the application.
+# Type: Integer
+# Default value: 4096
+#
+# universal_profiling_integration_buffer_size=4096
+
+# 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.
+# If unset, the value of the `java.io.tmpdir` system property will be used.
+#
+# This setting can not be changed at runtime. Changes require a restart of the application.
+# Type: String
+# Default value: 
+#
+# universal_profiling_integration_socket_dir=
+
 # Set to `true` to make the agent create spans for method executions based on
 # https://github.com/jvm-profiling-tools/async-profiler[async-profiler], a sampling aka statistical profiler.
 #