**Build scan:**
https://gradle-enterprise.elastic.co/s/6xjhol7yd4dge/tests/:x-pack:plugin:watcher:internalClusterTest/org.elasticsearch.xpack.watcher.transport.action.activate.ActivateWatchTests/testDeactivateAndActivate

**Reproduction line:**
`./gradlew ':x-pack:plugin:watcher:internalClusterTest' --tests "org.elasticsearch.xpack.watcher.transport.action.activate.ActivateWatchTests.testDeactivateAndActivate" -Dtests.seed=C05B271BF15F2E7C -Dtests.locale=ar-IQ -Dtests.timezone=America/Iqaluit -Druntime.java=17`

**Applicable branches:**
master

**Reproduces locally?:**
Didn't try

**Failure history:**
https://gradle-enterprise.elastic.co/scans/tests?tests.container=org.elasticsearch.xpack.watcher.transport.action.activate.ActivateWatchTests&tests.test=testDeactivateAndActivate

**Failure excerpt:**
```
java.lang.IllegalArgumentException: index [.ds-.watcher-history-14-2022.01.19-000001] is the write index for data stream [.watcher-history-14] and cannot be deleted

  at __randomizedtesting.SeedInfo.seed([C05B271BF15F2E7C:60035FE63E06F800]:0)
  at org.elasticsearch.cluster.metadata.MetadataDeleteIndexService.deleteIndices(MetadataDeleteIndexService.java:92)
  at org.elasticsearch.cluster.metadata.MetadataDeleteIndexService$1.execute(MetadataDeleteIndexService.java:69)
  at org.elasticsearch.cluster.ClusterStateTaskExecutor$1.execute(ClusterStateTaskExecutor.java:161)
  at org.elasticsearch.cluster.service.MasterService.executeTasks(MasterService.java:817)
  at org.elasticsearch.cluster.service.MasterService.calculateTaskOutputs(MasterService.java:392)
  at org.elasticsearch.cluster.service.MasterService.runTasks(MasterService.java:232)
  at org.elasticsearch.cluster.service.MasterService$Batcher.run(MasterService.java:164)
  at org.elasticsearch.cluster.service.TaskBatcher.runIfNotProcessed(TaskBatcher.java:152)
  at org.elasticsearch.cluster.service.TaskBatcher$BatchedTask.run(TaskBatcher.java:208)
  at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingRunnable.run(ThreadContext.java:716)
  at org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.runAndClean(PrioritizedEsThreadPoolExecutor.java:260)
  at org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.run(PrioritizedEsThreadPoolExecutor.java:223)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
  at java.lang.Thread.run(Thread.java:833)

```
x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/transport/action/activate/ActivateWatchTests.java
activateWatchResponse = new ActivateWatchRequestBuilder(client(), "_id", true).get();
        assertThat(activateWatchResponse, notNullValue());
        assertThat(activateWatchResponse.getStatus().state().isActive(), is(true));
        getWatchResponse = new GetWatchRequestBuilder(client(), "_id").get();
        assertThat(getWatchResponse, notNullValue());
        assertThat(getWatchResponse.getStatus().state().isActive(), is(true));
        refresh();
        assertBusy(() -> {
            long count2 = docCount(".watcher-history*", matchAllQuery());
            assertThat(count2, greaterThan(count1));
        });
    }
    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/82797")
    public void testLoadWatchWithoutAState() throws Exception {
        PutWatchResponse putWatchResponse = new PutWatchRequestBuilder(client()).setId("_id")
            .setSource(
                watchBuilder().trigger(schedule(cron("0 0 0 1 1 ? 2050"))) 
                    .input(simpleInput("foo", "bar"))
                    .addAction("_a1", indexAction("actions"))
                    .defaultThrottlePeriod(new TimeValue(0, TimeUnit.SECONDS))
            )
            .get();
        assertThat(putWatchResponse.isCreated(), is(true));
        GetWatchResponse getWatchResponse = new GetWatchRequestBuilder(client(), "_id").get();
        assertThat(getWatchResponse, notNullValue());
        assertThat(getWatchResponse.getStatus().state().isActive(), is(true));
        GetResponse getResponse = client().prepareGet().setIndex(".watches").setId("_id").get();
        XContentSource source = new XContentSource(getResponse.getSourceAsBytesRef(), XContentType.JSON);
        Set<String> filters = Sets.newHashSet(
            "trigger.**",