Skip to content

Commit c58dbb2

Browse files
gmarouligmarouli
authored
[8.x] Move the failure store enable flag into the data stream options (#113176) (#113799)
* Move the failure store enable flag into the data stream options (#113176) * Remove unsupported java 21 features
1 parent a3f6ba6 commit c58dbb2

File tree

13 files changed

+149
-77
lines changed

13 files changed

+149
-77
lines changed

‎modules/data-streams/src/test/java/org/elasticsearch/datastreams/action/GetDataStreamsResponseTests.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
import org.elasticsearch.cluster.health.ClusterHealthStatus;
1414
import org.elasticsearch.cluster.metadata.DataStream;
1515
import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
16+
import org.elasticsearch.cluster.metadata.DataStreamOptions;
1617
import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
1718
import org.elasticsearch.common.UUIDs;
1819
import org.elasticsearch.common.bytes.BytesReference;
@@ -83,7 +84,7 @@ public void testResponseIlmAndDataStreamLifecycleRepresentation() throws Excepti
8384
.setAllowCustomRouting(true)
8485
.setIndexMode(IndexMode.STANDARD)
8586
.setLifecycle(new DataStreamLifecycle())
86-
.setFailureStoreEnabled(true)
87+
.setDataStreamOptions(DataStreamOptions.FAILURE_STORE_ENABLED)
8788
.setFailureIndices(DataStream.DataStreamIndices.failureIndicesBuilder(failureStores).build())
8889
.build();
8990

@@ -186,7 +187,7 @@ public void testResponseIlmAndDataStreamLifecycleRepresentation() throws Excepti
186187
.setAllowCustomRouting(true)
187188
.setIndexMode(IndexMode.STANDARD)
188189
.setLifecycle(new DataStreamLifecycle(null, null, false))
189-
.setFailureStoreEnabled(true)
190+
.setDataStreamOptions(DataStreamOptions.FAILURE_STORE_ENABLED)
190191
.setFailureIndices(DataStream.DataStreamIndices.failureIndicesBuilder(failureStores).build())
191192
.build();
192193

‎modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
4343
import org.elasticsearch.cluster.metadata.DataStreamLifecycle.Downsampling;
4444
import org.elasticsearch.cluster.metadata.DataStreamLifecycle.Downsampling.Round;
45+
import org.elasticsearch.cluster.metadata.DataStreamOptions;
4546
import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
4647
import org.elasticsearch.cluster.metadata.IndexAbstraction;
4748
import org.elasticsearch.cluster.metadata.IndexGraveyard;
@@ -1492,6 +1493,13 @@ public void testTargetIndices() {
14921493
String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
14931494
int numBackingIndices = 3;
14941495
int numFailureIndices = 2;
1496+
int mutationBranch = randomIntBetween(0, 2);
1497+
DataStreamOptions dataStreamOptions = switch (mutationBranch) {
1498+
case 0 -> DataStreamOptions.EMPTY;
1499+
case 1 -> DataStreamOptions.FAILURE_STORE_ENABLED;
1500+
case 2 -> DataStreamOptions.FAILURE_STORE_DISABLED;
1501+
default -> throw new IllegalStateException("Unexpected value: " + mutationBranch);
1502+
};
14951503
Metadata.Builder builder = Metadata.builder();
14961504
DataStream dataStream = createDataStream(
14971505
builder,
@@ -1501,7 +1509,7 @@ public void testTargetIndices() {
15011509
settings(IndexVersion.current()),
15021510
new DataStreamLifecycle(),
15031511
now
1504-
).copy().setFailureStoreEnabled(randomBoolean()).build(); // failure store is managed even when disabled
1512+
).copy().setDataStreamOptions(dataStreamOptions).build(); // failure store is managed even when disabled
15051513
builder.put(dataStream);
15061514
Metadata metadata = builder.build();
15071515
Set<Index> indicesToExclude = Set.of(dataStream.getIndices().get(0), dataStream.getFailureIndices().getIndices().get(0));
@@ -1533,7 +1541,7 @@ public void testFailureStoreIsManagedEvenWhenDisabled() {
15331541
settings(IndexVersion.current()),
15341542
DataStreamLifecycle.newBuilder().dataRetention(0).build(),
15351543
now
1536-
).copy().setFailureStoreEnabled(false).build(); // failure store is managed even when it is disabled
1544+
).copy().setDataStreamOptions(DataStreamOptions.FAILURE_STORE_DISABLED).build(); // failure store is managed even when disabled
15371545
builder.put(dataStream);
15381546

15391547
ClusterState state = ClusterState.builder(ClusterName.DEFAULT).metadata(builder).build();

‎server/src/main/java/org/elasticsearch/TransportVersions.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -227,6 +227,7 @@ static TransportVersion def(int id) {
227227
public static final TransportVersion ML_INFERENCE_CHUNKING_SETTINGS = def(8_751_00_0);
228228
public static final TransportVersion SEMANTIC_QUERY_INNER_HITS = def(8_752_00_0);
229229
public static final TransportVersion RETAIN_ILM_STEP_INFO = def(8_753_00_0);
230+
public static final TransportVersion ADD_DATA_STREAM_OPTIONS = def(8_754_00_0);
230231

231232
/*
232233
* STOP! READ THIS FIRST! No, really,

‎server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java‎

Lines changed: 62 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,7 @@ public static boolean isFailureStoreFeatureFlagEnabled() {
112112
private final IndexMode indexMode;
113113
@Nullable
114114
private final DataStreamLifecycle lifecycle;
115-
private final boolean failureStoreEnabled;
115+
private final DataStreamOptions dataStreamOptions;
116116

117117
private final DataStreamIndices backingIndices;
118118
private final DataStreamIndices failureIndices;
@@ -128,7 +128,7 @@ public DataStream(
128128
boolean allowCustomRouting,
129129
IndexMode indexMode,
130130
DataStreamLifecycle lifecycle,
131-
boolean failureStoreEnabled,
131+
@Nullable DataStreamOptions dataStreamOptions,
132132
List<Index> failureIndices,
133133
boolean rolloverOnWrite,
134134
@Nullable DataStreamAutoShardingEvent autoShardingEvent
@@ -144,7 +144,7 @@ public DataStream(
144144
allowCustomRouting,
145145
indexMode,
146146
lifecycle,
147-
failureStoreEnabled,
147+
dataStreamOptions,
148148
new DataStreamIndices(BACKING_INDEX_PREFIX, List.copyOf(indices), rolloverOnWrite, autoShardingEvent),
149149
new DataStreamIndices(FAILURE_STORE_PREFIX, List.copyOf(failureIndices), false, null)
150150
);
@@ -162,7 +162,7 @@ public DataStream(
162162
boolean allowCustomRouting,
163163
IndexMode indexMode,
164164
DataStreamLifecycle lifecycle,
165-
boolean failureStoreEnabled,
165+
DataStreamOptions dataStreamOptions,
166166
DataStreamIndices backingIndices,
167167
DataStreamIndices failureIndices
168168
) {
@@ -177,7 +177,7 @@ public DataStream(
177177
this.allowCustomRouting = allowCustomRouting;
178178
this.indexMode = indexMode;
179179
this.lifecycle = lifecycle;
180-
this.failureStoreEnabled = failureStoreEnabled;
180+
this.dataStreamOptions = dataStreamOptions == null ? DataStreamOptions.EMPTY : dataStreamOptions;
181181
assert backingIndices.indices.isEmpty() == false;
182182
assert replicated == false || (backingIndices.rolloverOnWrite == false && failureIndices.rolloverOnWrite == false)
183183
: "replicated data streams cannot be marked for lazy rollover";
@@ -198,9 +198,11 @@ public static DataStream read(StreamInput in) throws IOException {
198198
var lifecycle = in.getTransportVersion().onOrAfter(TransportVersions.V_8_9_X)
199199
? in.readOptionalWriteable(DataStreamLifecycle::new)
200200
: null;
201-
var failureStoreEnabled = in.getTransportVersion().onOrAfter(DataStream.ADDED_FAILURE_STORE_TRANSPORT_VERSION)
202-
? in.readBoolean()
203-
: false;
201+
// This boolean flag has been moved in data stream options
202+
var failureStoreEnabled = in.getTransportVersion()
203+
.between(DataStream.ADDED_FAILURE_STORE_TRANSPORT_VERSION, TransportVersions.ADD_DATA_STREAM_OPTIONS)
204+
? in.readBoolean()
205+
: false;
204206
var failureIndices = in.getTransportVersion().onOrAfter(DataStream.ADDED_FAILURE_STORE_TRANSPORT_VERSION)
205207
? readIndices(in)
206208
: List.<Index>of();
@@ -213,6 +215,14 @@ public static DataStream read(StreamInput in) throws IOException {
213215
failureIndicesBuilder.setRolloverOnWrite(in.readBoolean())
214216
.setAutoShardingEvent(in.readOptionalWriteable(DataStreamAutoShardingEvent::new));
215217
}
218+
DataStreamOptions dataStreamOptions;
219+
if (in.getTransportVersion().onOrAfter(TransportVersions.ADD_DATA_STREAM_OPTIONS)) {
220+
dataStreamOptions = in.readOptionalWriteable(DataStreamOptions::read);
221+
} else {
222+
// We cannot distinguish if failure store was explicitly disabled or not. Given that failure store
223+
// is still behind a feature flag in previous version we use the default value instead of explicitly disabling it.
224+
dataStreamOptions = failureStoreEnabled ? DataStreamOptions.FAILURE_STORE_ENABLED : null;
225+
}
216226
return new DataStream(
217227
name,
218228
generation,
@@ -224,7 +234,7 @@ public static DataStream read(StreamInput in) throws IOException {
224234
allowCustomRouting,
225235
indexMode,
226236
lifecycle,
227-
failureStoreEnabled,
237+
dataStreamOptions,
228238
backingIndicesBuilder.build(),
229239
failureIndicesBuilder.build()
230240
);
@@ -274,6 +284,10 @@ public boolean isFailureStoreIndex(String indexName) {
274284
return failureIndices.containsIndex(indexName);
275285
}
276286

287+
public DataStreamOptions getDataStreamOptions() {
288+
return dataStreamOptions;
289+
}
290+
277291
public boolean rolloverOnWrite() {
278292
return backingIndices.rolloverOnWrite;
279293
}
@@ -406,13 +420,12 @@ public boolean isAllowCustomRouting() {
406420
}
407421

408422
/**
409-
* Determines if this data stream should persist ingest pipeline and mapping failures from bulk requests to a locally
410-
* configured failure store.
411-
*
412-
* @return Whether this data stream should store ingestion failures.
423+
* Determines if this data stream has its failure store enabled or not. Currently, the failure store
424+
* is enabled only when a user has explicitly requested it.
425+
* @return true, if the user has explicitly enabled the failure store.
413426
*/
414427
public boolean isFailureStoreEnabled() {
415-
return failureStoreEnabled;
428+
return dataStreamOptions.failureStore() != null && dataStreamOptions.failureStore().isExplicitlyEnabled();
416429
}
417430

418431
@Nullable
@@ -1063,8 +1076,11 @@ public void writeTo(StreamOutput out) throws IOException {
10631076
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_9_X)) {
10641077
out.writeOptionalWriteable(lifecycle);
10651078
}
1079+
if (out.getTransportVersion()
1080+
.between(DataStream.ADDED_FAILURE_STORE_TRANSPORT_VERSION, TransportVersions.ADD_DATA_STREAM_OPTIONS)) {
1081+
out.writeBoolean(isFailureStoreEnabled());
1082+
}
10661083
if (out.getTransportVersion().onOrAfter(DataStream.ADDED_FAILURE_STORE_TRANSPORT_VERSION)) {
1067-
out.writeBoolean(failureStoreEnabled);
10681084
out.writeCollection(failureIndices.indices);
10691085
}
10701086
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_13_0)) {
@@ -1077,6 +1093,9 @@ public void writeTo(StreamOutput out) throws IOException {
10771093
out.writeBoolean(failureIndices.rolloverOnWrite);
10781094
out.writeOptionalWriteable(failureIndices.autoShardingEvent);
10791095
}
1096+
if (out.getTransportVersion().onOrAfter(TransportVersions.ADD_DATA_STREAM_OPTIONS)) {
1097+
out.writeOptionalWriteable(dataStreamOptions.isEmpty() ? null : dataStreamOptions);
1098+
}
10801099
}
10811100

10821101
public static final ParseField NAME_FIELD = new ParseField("name");
@@ -1096,6 +1115,7 @@ public void writeTo(StreamOutput out) throws IOException {
10961115
public static final ParseField AUTO_SHARDING_FIELD = new ParseField("auto_sharding");
10971116
public static final ParseField FAILURE_ROLLOVER_ON_WRITE_FIELD = new ParseField("failure_rollover_on_write");
10981117
public static final ParseField FAILURE_AUTO_SHARDING_FIELD = new ParseField("failure_auto_sharding");
1118+
public static final ParseField DATA_STREAM_OPTIONS_FIELD = new ParseField("options");
10991119

11001120
@SuppressWarnings("unchecked")
11011121
private static final ConstructingObjectParser<DataStream, Void> PARSER = new ConstructingObjectParser<>("data_stream", args -> {
@@ -1110,6 +1130,16 @@ public void writeTo(StreamOutput out) throws IOException {
11101130
(DataStreamAutoShardingEvent) args[15]
11111131
)
11121132
: new DataStreamIndices(FAILURE_STORE_PREFIX, List.of(), false, null);
1133+
// We cannot distinguish if failure store was explicitly disabled or not. Given that failure store
1134+
// is still behind a feature flag in previous version we use the default value instead of explicitly disabling it.
1135+
DataStreamOptions dataStreamOptions = DataStreamOptions.EMPTY;
1136+
if (DataStream.isFailureStoreFeatureFlagEnabled()) {
1137+
if (args[16] != null) {
1138+
dataStreamOptions = (DataStreamOptions) args[16];
1139+
} else if (failureStoreEnabled) {
1140+
dataStreamOptions = DataStreamOptions.FAILURE_STORE_ENABLED;
1141+
}
1142+
}
11131143
return new DataStream(
11141144
(String) args[0],
11151145
(Long) args[2],
@@ -1121,7 +1151,7 @@ public void writeTo(StreamOutput out) throws IOException {
11211151
args[7] != null && (boolean) args[7],
11221152
args[8] != null ? IndexMode.fromString((String) args[8]) : null,
11231153
(DataStreamLifecycle) args[9],
1124-
failureStoreEnabled,
1154+
dataStreamOptions,
11251155
new DataStreamIndices(
11261156
BACKING_INDEX_PREFIX,
11271157
(List<Index>) args[1],
@@ -1171,6 +1201,11 @@ public void writeTo(StreamOutput out) throws IOException {
11711201
(p, c) -> DataStreamAutoShardingEvent.fromXContent(p),
11721202
FAILURE_AUTO_SHARDING_FIELD
11731203
);
1204+
PARSER.declareObject(
1205+
ConstructingObjectParser.optionalConstructorArg(),
1206+
(p, c) -> DataStreamOptions.fromXContent(p),
1207+
DATA_STREAM_OPTIONS_FIELD
1208+
);
11741209
}
11751210
}
11761211

@@ -1208,7 +1243,6 @@ public XContentBuilder toXContent(
12081243
builder.field(SYSTEM_FIELD.getPreferredName(), system);
12091244
builder.field(ALLOW_CUSTOM_ROUTING.getPreferredName(), allowCustomRouting);
12101245
if (DataStream.isFailureStoreFeatureFlagEnabled()) {
1211-
builder.field(FAILURE_STORE_FIELD.getPreferredName(), failureStoreEnabled);
12121246
if (failureIndices.indices.isEmpty() == false) {
12131247
builder.xContentList(FAILURE_INDICES_FIELD.getPreferredName(), failureIndices.indices);
12141248
}
@@ -1218,6 +1252,10 @@ public XContentBuilder toXContent(
12181252
failureIndices.autoShardingEvent.toXContent(builder, params);
12191253
builder.endObject();
12201254
}
1255+
if (dataStreamOptions.isEmpty() == false) {
1256+
builder.field(DATA_STREAM_OPTIONS_FIELD.getPreferredName());
1257+
dataStreamOptions.toXContent(builder, params);
1258+
}
12211259
}
12221260
if (indexMode != null) {
12231261
builder.field(INDEX_MODE.getPreferredName(), indexMode);
@@ -1250,7 +1288,7 @@ public boolean equals(Object o) {
12501288
&& allowCustomRouting == that.allowCustomRouting
12511289
&& indexMode == that.indexMode
12521290
&& Objects.equals(lifecycle, that.lifecycle)
1253-
&& failureStoreEnabled == that.failureStoreEnabled
1291+
&& Objects.equals(dataStreamOptions, that.dataStreamOptions)
12541292
&& Objects.equals(backingIndices, that.backingIndices)
12551293
&& Objects.equals(failureIndices, that.failureIndices);
12561294
}
@@ -1267,7 +1305,7 @@ public int hashCode() {
12671305
allowCustomRouting,
12681306
indexMode,
12691307
lifecycle,
1270-
failureStoreEnabled,
1308+
dataStreamOptions,
12711309
backingIndices,
12721310
failureIndices
12731311
);
@@ -1580,7 +1618,7 @@ public static class Builder {
15801618
private IndexMode indexMode = null;
15811619
@Nullable
15821620
private DataStreamLifecycle lifecycle = null;
1583-
private boolean failureStoreEnabled = false;
1621+
private DataStreamOptions dataStreamOptions = DataStreamOptions.EMPTY;
15841622
private DataStreamIndices backingIndices;
15851623
private DataStreamIndices failureIndices = DataStreamIndices.failureIndicesBuilder(List.of()).build();
15861624

@@ -1605,7 +1643,7 @@ private Builder(DataStream dataStream) {
16051643
allowCustomRouting = dataStream.allowCustomRouting;
16061644
indexMode = dataStream.indexMode;
16071645
lifecycle = dataStream.lifecycle;
1608-
failureStoreEnabled = dataStream.failureStoreEnabled;
1646+
dataStreamOptions = dataStream.dataStreamOptions;
16091647
backingIndices = dataStream.backingIndices;
16101648
failureIndices = dataStream.failureIndices;
16111649
}
@@ -1660,8 +1698,8 @@ public Builder setLifecycle(DataStreamLifecycle lifecycle) {
16601698
return this;
16611699
}
16621700

1663-
public Builder setFailureStoreEnabled(boolean failureStoreEnabled) {
1664-
this.failureStoreEnabled = failureStoreEnabled;
1701+
public Builder setDataStreamOptions(DataStreamOptions dataStreamOptions) {
1702+
this.dataStreamOptions = dataStreamOptions;
16651703
return this;
16661704
}
16671705

@@ -1697,7 +1735,7 @@ public DataStream build() {
16971735
allowCustomRouting,
16981736
indexMode,
16991737
lifecycle,
1700-
failureStoreEnabled,
1738+
dataStreamOptions,
17011739
backingIndices,
17021740
failureIndices
17031741
);

0 commit comments

Comments
 (0)