**Found in version**
- Observed in 8.2.3, but might exist longer

**Steps to reproduce**
On a multi node cloud cluster (might also work on a local multi node cluster):
- Create a transform with a retention policy
- Update the transform and remove the retention policy

**Expected result**
- The retention policy is removed

**Actual result**
- This sometimes results in an error:
```
An error occurred calling the API endpoint to update transforms.
Bad Request: [illegal_argument_exception: [illegal_argument_exception] Reason: 
Unknown NamedWriteable [org.elasticsearch.xpack.core.transform.transforms.RetentionPolicyConfig][null_retention_policy]]:
Unknown NamedWriteable [org.elasticsearch.xpack.core.transform.transforms.RetentionPolicyConfig][null_retention_policy]
```

**Additional information**
- Quote from slack discussion with @benwtrent:
  > It wouldn’t happen if you hit a node that can execute the update immediately without serializing to a different node
x-pack/plugin/transform/qa/multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformIT.java
Map<String, SingleGroupSource> groups = new HashMap<>();
        groups.put("by-user", new TermsGroupSource("user_id", null, false));
        AggregatorFactories.Builder aggs = AggregatorFactories.builder()
            .addAggregator(AggregationBuilders.avg("review_score").field("stars"))
            .addAggregator(AggregationBuilders.max("timestamp").field("timestamp"));
        String id = "transform-to-update";
        String dest = "reviews-by-user-business-day-to-update";
        TransformConfig config = createTransformConfigBuilder(id, dest, QueryConfig.matchAll(), indexName).setPivotConfig(
            createPivotConfig(groups, aggs)
        ).setSyncConfig(new TimeSyncConfig("timestamp", TimeValue.timeValueSeconds(1))).build();
        putTransform(id, Strings.toString(config), RequestOptions.DEFAULT);
        startTransform(config.getId(), RequestOptions.DEFAULT);
        waitUntilCheckpoint(config.getId(), 1L);
        assertThat(getTransformState(config.getId()), oneOf("started", "indexing"));
        int docsIndexed = (Integer) XContentMapValues.extractValue("stats.documents_indexed", getTransformStats(config.getId()));
        var storedConfig = getTransform(config.getId());
        assertThat(storedConfig.get("version"), equalTo(Version.CURRENT.toString()));
        Instant now = Instant.now();
        long createTime = (long) storedConfig.get("create_time");
        assertTrue("[create_time] is not before current time", Instant.ofEpochMilli(createTime).isBefore(now));
        String pipelineId = "add_forty_two";
        final XContentBuilder pipelineBuilder = jsonBuilder().startObject()
            .startArray("processors")
            .startObject()
            .startObject("set")
            .field("field", "static_forty_two")
            .field("value", 42)
            .endObject()
            .endObject()
            .endArray()