### Elasticsearch Version

8.5.0-SNAPSHOT

### Installed Plugins

_No response_

### Java Version

_bundled_

### OS Version

any

### Problem Description

[This PR to kibana](https://github.com/elastic/kibana/pull/140121) which adds APM instrumentation to any Elasticsearch instance Kibana starts (integration tests/e2e tests/local dev) is consistently failing on a ML e2e tests. 


[The CI logs](https://s3.amazonaws.com/buildkiteartifacts.com/e0f3970e-3a75-4621-919f-e6c773e2bb12/0fda5127-f57f-42fb-8e5a-146b3d535916/01831401-63aa-4798-9f6a-d5710e0c8b3b/01831403-7c1f-4002-8ff7-cd53598a2dfc/target/test_failures/01831403-7c1f-4002-8ff7-cd53598a2dfc_9a8aa981a5d9e402eed13abb59b26b55.html?response-content-type=text%2Fhtml&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=ASIAQPCP3C7LWDYERT55%2F20220907%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20220907T080644Z&X-Amz-Expires=600&X-Amz-Security-Token=IQoJb3JpZ2luX2VjEHIaCXVzLWVhc3QtMSJHMEUCIQC3cOzS8h6eGVBMR8qhJboliLOE0ySMxtKWvkbB%2B%2BqIigIgXW1eVdfvDEciqG2qL8y%2FjOVw6Qw11vYJAz0rhqcpD%2FYq1QQI%2B%2F%2F%2F%2F%2F%2F%2F%2F%2F%2F%2FARAAGgwwMzIzNzk3MDUzMDMiDGQVJH6UMqM%2BMFucISqpBPd%2BXccbjndUxN%2BM7cDOwSXUl%2BxseIezmRDbMa29p%2FnKlgLv3d30gfzYanfdkbvW7d4zc5rN4eEpstjIV7OygeldBfVZmecX2dntVu%2FUcuczI2PoLi5TjKBXP0RBhI%2BJUGRkQmj1hdb%2FWiRhE7l9UWYu%2BIpwaLa1OeoBao32kzf0pD%2FZQnUcPSwf%2FjVLwM3wqxgosbivhajfhSXc6JFBq77DnxSwY6uXhEGh8wJEEyVG175ZYD9nXaklXZneau4M1EHK7pGMwq3Y5Ta8Qv20PXNApB5gcgVYSxZ4t%2BMWvxwaSRrYSd16njlTox6cwOcABF0Ls81DDXrhUFBJWFhol7e0%2BtR%2BLvUTDaqUyhAGICArQSMWajN8lOPD1luWS8aw9Zn6GegRQmkF%2B1TlPODdPN5763xa9M2W6uiSBDWQ3FcqylzoUJvwnVQCp%2Bsy94eLYic0Xs6JGBUP7v%2FQ%2FJ4KCQGFnXltjNSO%2F5FYKYqvtbSQo0YeyXlu6qQWmu3X2Dn8gfndDLMr06U9I23%2FAHmxZuJ%2BvqbzdQvEUh2aKQIWO%2FnoTslALEBcGdiBD%2BW7VENbyMEcfb2hMPdNAygELW9AOe5VMbR2ZPkL%2F4kovRYAidWbX2kLrfnDHEU2%2B%2BYJuOL%2FuVMM%2BC3yqQnvZUIIXovBzBcIJn85ZyCXQbq001QUOWKmFjm0N0ik04F8hRrdyaSEc8i4ntPXNHJJQI84b22jcf9yxaPHlIbWAScww%2FXfmAY6qQGoFK8HyIUxNF1flu9DS801cAW28vb3y22nWJiyYmDrmpPA596bE2apP%2F0wZI9xbPhRJz9bueKWgiGsxCWv%2FU1wj2pcHdK0EKy7AR4GR0sU%2FcNkgczl0vw3xXPpzKLgcHqiYdOqbyHLFU63eCXgGL3K0XezcaGXTVTw03pMQ6GyR7eiRs2DifLziKiVWH%2FLLseuCpjvfUrr5jB7xsFt94yHQ0DeiwRR%2F3Fw&X-Amz-SignedHeaders=host&X-Amz-Signature=95628f046b2741871ad36c8abd3eb4b9516be23a3842f6e16ec01ab916bdaa04) indicate that `apm.local.context` is being written more than once to the `ThreadContext` which it doesn't support. 

> [00:01:37]               │ info [o.e.c.s.ClusterApplierService] [ftr] failed to notify ClusterStateListener java.lang.IllegalArgumentException: value for key [apm.local.context] already present
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.common.util.concurrent.ThreadContext$ThreadContextStruct.putSingleHeader(ThreadContext.java:677)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.common.util.concurrent.ThreadContext$ThreadContextStruct.putTransient(ThreadContext.java:783)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.common.util.concurrent.ThreadContext.putTransient(ThreadContext.java:502)
[00:01:37]               │      	at org.elasticsearch.tracing.apm.APMTracer.updateThreadContext(APMTracer.java:184)
[00:01:37]               │      	at org.elasticsearch.tracing.apm.APMTracer.lambda$startTrace$1(APMTracer.java:176)
[00:01:37]               │      	at java.base/java.security.AccessController.doPrivileged(AccessController.java:318)
[00:01:37]               │      	at org.elasticsearch.tracing.apm.APMTracer.lambda$startTrace$2(APMTracer.java:161)
[00:01:37]               │      	at java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1708)
[00:01:37]               │      	at org.elasticsearch.tracing.apm.APMTracer.startTrace(APMTracer.java:161)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.tasks.TaskManager.startTrace(TaskManager.java:176)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.tasks.TaskManager.registerCancellableTask(TaskManager.java:237)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.tasks.TaskManager.register(TaskManager.java:157)
[00:01:37]               │      	at org.elasticsearch.xpack.ml.inference.assignment.TrainedModelAssignmentNodeService.prepareModelToLoad(TrainedModelAssignmentNodeService.java:463)
[00:01:37]               │      	at org.elasticsearch.xpack.ml.inference.assignment.TrainedModelAssignmentNodeService.clusterChanged(TrainedModelAssignmentNodeService.java:358)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.cluster.service.ClusterApplierService.callClusterStateListener(ClusterApplierService.java:558)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.cluster.service.ClusterApplierService.callClusterStateListeners(ClusterApplierService.java:544)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.cluster.service.ClusterApplierService.applyChanges(ClusterApplierService.java:504)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.cluster.service.ClusterApplierService.runTask(ClusterApplierService.java:428)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.cluster.service.ClusterApplierService$UpdateTask.run(ClusterApplierService.java:154)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingRunnable.run(ThreadContext.java:825)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.runAndClean(PrioritizedEsThreadPoolExecutor.java:257)
[00:01:37]               │      	at org.elasticsearch.server@8.5.0-SNAPSHOT/org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.run(PrioritizedEsThreadPoolExecutor.java:223)
[00:01:37]               │      	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
[00:01:37]               │      	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
[00:01:37]               │      	at java.base/java.lang.Thread.run(Thread.java:833)


This by itself is the bug report.

I want to highlight it later _seems_ to cause ML parsing issues:

> info [o.e.x.m.i.l.ModelLoadingService] [ftr] [dfa_classification_model_n_14] failed to load model definition org.elasticsearch.xcontent.XContentParseException: [8:5] [inference_model_definition] failed to parse field [trained_model]

Which appear to consistently fail the e2e tests on the PR adding the instrumentation but not on kibana's `main` branch. 
Implying a very surprising correlation between the two Elasticsearch Exceptions being logged. 


### Steps to Reproduce

run tests https://github.com/elastic/kibana/pull/140121

### Logs (if relevant)

_No response_
server/src/test/java/org/elasticsearch/tasks/TaskManagerTests.java
}
        assertEquals(600000L, total);
    }
    public void testTrackingChannelTask() throws Exception {
        final TaskManager taskManager = new TaskManager(Settings.EMPTY, threadPool, Set.of(), Tracer.NOOP);
        Set<Task> cancelledTasks = ConcurrentCollections.newConcurrentSet();
        final var transportServiceMock = mock(TransportService.class);
        when(transportServiceMock.getThreadPool()).thenReturn(threadPool);
        taskManager.setTaskCancellationService(new TaskCancellationService(transportServiceMock) {
            @Override
            void cancelTaskAndDescendants(CancellableTask task, String reason, boolean waitForCompletion, ActionListener<Void> listener) {
                assertThat(reason, equalTo("channel was closed"));
                assertFalse(waitForCompletion);
                assertTrue("task [" + task + "] was cancelled already", cancelledTasks.add(task));
            }
        });
        Map<TcpChannel, Set<Task>> pendingTasks = new HashMap<>();
        Set<Task> expectedCancelledTasks = new HashSet<>();
        FakeTcpChannel[] channels = new FakeTcpChannel[randomIntBetween(1, 10)];
        List<Releasable> stopTrackingTasks = new ArrayList<>();
        for (int i = 0; i < channels.length; i++) {
            channels[i] = new SingleThreadedTcpChannel();
        }
        int iterations = randomIntBetween(1, 200);
        for (int i = 0; i < iterations; i++) {
            final List<Releasable> subset = randomSubsetOf(stopTrackingTasks);
            stopTrackingTasks.removeAll(subset);
            Releasables.close(subset);
            final FakeTcpChannel channel = randomFrom(channels);
            final Task task = taskManager.register("transport", "test", new CancellableRequest(Integer.toString(i)));