Randomly run CCR tests with _source disabled (#49922)

Makes sure that CCR also properly works with _source disabled.

Changes one exception in LuceneChangesSnapshot as the case of missing _recovery_source
because of a missing lease was not properly properly bubbled up to CCR (testIndexFallBehind
was failing).

Author: ywelsch

server/src/main/java/org/elasticsearch/index/engine/LuceneChangesSnapshot.java

@@ -255,7 +255,7 @@ private Translog.Operation readDocAsOp(int docIndex) throws IOException {
                     // TODO: Callers should ask for the range that source should be retained. Thus we should always
                     // check for the existence source once we make peer-recovery to send ops after the local checkpoint.
                     if (requiredFullRange) {
-                        throw new IllegalStateException("source not found for seqno=" + seqNo +
+                        throw new MissingHistoryOperationsException("source not found for seqno=" + seqNo +
                             " from_seqno=" + fromSeqNo + " to_seqno=" + toSeqNo);
                     } else {
                         skippedOperations++;

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrIntegTestCase.java

@@ -405,6 +405,14 @@ protected void ensureNoCcrTasks() throws Exception {
         }, 30, TimeUnit.SECONDS);
     }
 
+
+    @Before
+    public void setupSourceEnabledOrDisabled() {
+        sourceEnabled = randomBoolean();
+    }
+
+    protected boolean sourceEnabled;
+
     protected String getIndexSettings(final int numberOfShards, final int numberOfReplicas,
                                     final Map<String, String> additionalIndexSettings) throws IOException {
         final String settings;
@@ -435,6 +443,11 @@ protected String getIndexSettings(final int numberOfShards, final int numberOfRe
                             builder.endObject();
                         }
                         builder.endObject();
+                        if (sourceEnabled == false) {
+                            builder.startObject("_source");
+                            builder.field("enabled", false);
+                            builder.endObject();
+                        }
                     }
                     builder.endObject();
                 }

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/CcrRepositoryIT.java

@@ -446,8 +446,10 @@ public void testFollowerMappingIsUpdated() throws IOException {
     private void assertExpectedDocument(String followerIndex, final int value) {
         final GetResponse getResponse = followerClient().prepareGet(followerIndex, Integer.toString(value)).get();
         assertTrue("Doc with id [" + value + "] is missing", getResponse.isExists());
-        assertTrue((getResponse.getSource().containsKey("f")));
-        assertThat(getResponse.getSource().get("f"), equalTo(value));
+        if (sourceEnabled) {
+            assertTrue((getResponse.getSource().containsKey("f")));
+            assertThat(getResponse.getSource().get("f"), equalTo(value));
+        }
     }
 
 }

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/CcrRetentionLeaseIT.java

@@ -1075,8 +1075,10 @@ private String getRetentionLeaseId(String followerIndex, String followerUUID, St
     private void assertExpectedDocument(final String followerIndex, final int value) {
         final GetResponse getResponse = followerClient().prepareGet(followerIndex, Integer.toString(value)).get();
         assertTrue("doc with id [" + value + "] is missing", getResponse.isExists());
-        assertTrue((getResponse.getSource().containsKey("f")));
-        assertThat(getResponse.getSource().get("f"), equalTo(value));
+        if (sourceEnabled) {
+            assertTrue((getResponse.getSource().containsKey("f")));
+            assertThat(getResponse.getSource().get("f"), equalTo(value));
+        }
     }
 
 }

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/IndexFollowingIT.java

@@ -1519,8 +1519,10 @@ private CheckedRunnable<Exception> assertExpectedDocumentRunnable(final int key,
         return () -> {
             final GetResponse getResponse = followerClient().prepareGet("index2", Integer.toString(key)).get();
             assertTrue("Doc with id [" + key + "] is missing", getResponse.isExists());
-            assertTrue((getResponse.getSource().containsKey("f")));
-            assertThat(getResponse.getSource().get("f"), equalTo(value));
+            if (sourceEnabled) {
+                assertTrue((getResponse.getSource().containsKey("f")));
+                assertThat(getResponse.getSource().get("f"), equalTo(value));
+            }
         };
     }