Skip to content

Commit f5a5b86

Browse files
samxbrsamxbr
committed
Address comments
1 parent 69240ed commit f5a5b86

File tree

3 files changed

+20
-39
lines changed

3 files changed

+20
-39
lines changed

‎modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/GeoIpDownloaderTaskExecutor.java‎

Lines changed: 16 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
import java.util.Objects;
5454
import java.util.Set;
5555
import java.util.concurrent.ConcurrentHashMap;
56+
import java.util.concurrent.atomic.AtomicBoolean;
5657

5758
import static org.elasticsearch.ingest.geoip.GeoIpDownloader.DATABASES_INDEX;
5859
import static org.elasticsearch.ingest.geoip.GeoIpDownloader.GEOIP_DOWNLOADER;
@@ -104,7 +105,7 @@ public final class GeoIpDownloaderTaskExecutor extends PersistentTasksExecutor<G
104105
private volatile boolean eagerDownload;
105106

106107
private final ConcurrentHashMap<ProjectId, Boolean> atLeastOneGeoipProcessorByProject = new ConcurrentHashMap<>();
107-
private final ConcurrentHashMap<ProjectId, Boolean> taskIsBootstrappedByProject = new ConcurrentHashMap<>();
108+
private final ConcurrentHashMap<ProjectId, AtomicBoolean> taskIsBootstrappedByProject = new ConcurrentHashMap<>();
108109
private final ConcurrentHashMap<ProjectId, GeoIpDownloader> tasks = new ConcurrentHashMap<>();
109110
private final ProjectResolver projectResolver;
110111

@@ -133,24 +134,22 @@ public void init() {
133134

134135
@FixForMultiProject(description = "Should execute in the context of the current project after settings are project-aware")
135136
private void setEnabled(boolean enabled) {
136-
assert projectResolver.getProjectId() != null : "projectId must be set before enabling download";
137137
if (clusterService.state().nodes().isLocalNodeElectedMaster() == false) {
138138
// we should only start/stop task from single node, master is the best as it will go through it anyway
139139
return;
140140
}
141141
if (enabled) {
142-
startTask(projectResolver.getProjectId(), () -> {});
142+
startTask(ProjectId.DEFAULT, () -> {});
143143
} else {
144-
stopTask(projectResolver.getProjectId(), () -> {});
144+
stopTask(ProjectId.DEFAULT, () -> {});
145145
}
146146
}
147147

148148
@FixForMultiProject(description = "Should execute in the context of the current project after settings are project-aware")
149149
private void setEagerDownload(Boolean eagerDownload) {
150-
assert projectResolver.getProjectId() != null : "projectId must be set before setting eager download";
151150
if (Objects.equals(this.eagerDownload, eagerDownload) == false) {
152151
this.eagerDownload = eagerDownload;
153-
GeoIpDownloader currentDownloader = getTask(projectResolver.getProjectId());
152+
GeoIpDownloader currentDownloader = getTask(ProjectId.DEFAULT);
154153
if (currentDownloader != null && Objects.equals(eagerDownload, Boolean.TRUE)) {
155154
currentDownloader.requestReschedule();
156155
}
@@ -159,10 +158,9 @@ private void setEagerDownload(Boolean eagerDownload) {
159158

160159
@FixForMultiProject(description = "Should execute in the context of the current project after settings are project-aware")
161160
private void setPollInterval(TimeValue pollInterval) {
162-
assert projectResolver.getProjectId() != null : "projectId must be set before setting poll interval";
163161
if (Objects.equals(this.pollInterval, pollInterval) == false) {
164162
this.pollInterval = pollInterval;
165-
GeoIpDownloader currentDownloader = getTask(projectResolver.getProjectId());
163+
GeoIpDownloader currentDownloader = getTask(ProjectId.DEFAULT);
166164
if (currentDownloader != null) {
167165
currentDownloader.requestReschedule();
168166
}
@@ -209,6 +207,7 @@ protected GeoIpDownloader createTask(
209207
);
210208
}
211209

210+
@FixForMultiProject(description = "Make sure removed project tasks are cancelled: https://elasticco.atlassian.net/browse/ES-12054")
212211
@Override
213212
public void clusterChanged(ClusterChangedEvent event) {
214213
if (event.state().blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) {
@@ -226,21 +225,20 @@ public void clusterChanged(ClusterChangedEvent event) {
226225
return;
227226
}
228227

229-
for (var projectMetadataEntry : event.state().metadata().projects().entrySet()) {
230-
ProjectId projectId = projectMetadataEntry.getKey();
231-
ProjectMetadata projectMetadata = projectMetadataEntry.getValue();
228+
for (var projectMetadata : event.state().metadata().projects().values()) {
229+
ProjectId projectId = projectMetadata.id();
232230

233231
// bootstrap task once iff it is not already bootstrapped
234-
boolean taskIsBootstrapped = taskIsBootstrappedByProject.computeIfAbsent(projectId, k -> false);
235-
if (taskIsBootstrapped != true) {
236-
taskIsBootstrappedByProject.put(projectId, true);
237-
this.taskIsBootstrappedByProject.computeIfAbsent(projectId, k -> hasAtLeastOneGeoipProcessor(projectMetadata));
232+
AtomicBoolean taskIsBootstrapped = taskIsBootstrappedByProject.computeIfAbsent(projectId, k -> new AtomicBoolean(false));
233+
if (taskIsBootstrapped.getAndSet(true) == false) {
234+
this.taskIsBootstrappedByProject.computeIfAbsent(projectId,
235+
k -> new AtomicBoolean(hasAtLeastOneGeoipProcessor(projectMetadata)));
238236
if (ENABLED_SETTING.get(event.state().getMetadata().settings(), settings)) {
239237
logger.debug("Bootstrapping geoip downloader task for project [{}]", projectId);
240-
startTask(projectId, () -> taskIsBootstrappedByProject.put(projectId, false));
238+
startTask(projectId, () -> taskIsBootstrapped.set(false));
241239
} else {
242240
logger.debug("Stopping geoip downloader task for project [{}]", projectId);
243-
stopTask(projectId, () -> taskIsBootstrappedByProject.put(projectId, false));
241+
stopTask(projectId, () -> taskIsBootstrapped.set(false));
244242
}
245243
}
246244

@@ -430,8 +428,8 @@ private void stopTask(ProjectId projectId, Runnable onFailure) {
430428
MasterNodeRequest.INFINITE_MASTER_NODE_TIMEOUT,
431429
ActionListener.runAfter(listener, () -> {
432430
IndexAbstraction databasesAbstraction = clusterService.state()
433-
.projectState(projectId)
434431
.metadata()
432+
.getProject(projectId)
435433
.getIndicesLookup()
436434
.get(DATABASES_INDEX);
437435
if (databasesAbstraction != null) {

‎modules/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpDownloaderTaskExecutorTests.java‎

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -271,17 +271,6 @@ private String getGeoIpProcessor(boolean downloadDatabaseOnPipelineCreation) thr
271271
}
272272
}
273273

274-
private ClusterState clusterStateWithIndex(Consumer<Settings.Builder> consumer, IngestMetadata ingestMetadata) {
275-
var builder = indexSettings(IndexVersion.current(), 1, 1);
276-
consumer.accept(builder);
277-
var indexMetadata = new IndexMetadata.Builder("index").settings(builder.build()).build();
278-
var project = ProjectMetadata.builder(Metadata.DEFAULT_PROJECT_ID)
279-
.putCustom(IngestMetadata.TYPE, ingestMetadata)
280-
.put(indexMetadata, false)
281-
.build();
282-
return ClusterState.builder(ClusterName.DEFAULT).putProjectMetadata(project).build();
283-
}
284-
285274
private ProjectMetadata projectMetadataWithIndex(Consumer<Settings.Builder> consumer, IngestMetadata ingestMetadata) {
286275
var builder = indexSettings(IndexVersion.current(), 1, 1);
287276
consumer.accept(builder);

‎server/src/main/java/org/elasticsearch/cluster/ClusterChangedEvent.java‎

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -169,17 +169,11 @@ public Set<String> changedCustomProjectMetadataSet() {
169169
* updated or removed between the previous and the current state
170170
*/
171171
public boolean customMetadataChanged(ProjectId projectId, String customMetadataType) {
172-
Set<String> changed = new HashSet<>();
173-
ProjectMetadata project = state.metadata().projects().get(projectId);
174172
ProjectMetadata previousProject = previousState.metadata().projects().get(projectId);
175-
if (previousProject != null && project != null) {
176-
changed.addAll(changedCustoms(project.customs(), previousProject.customs()));
177-
} else if (previousProject != null) {
178-
changed.addAll(previousProject.customs().keySet());
179-
} else if (project != null) {
180-
changed.addAll(project.customs().keySet());
181-
}
182-
return changed.contains(customMetadataType);
173+
ProjectMetadata project = state.metadata().projects().get(projectId);
174+
Object previousValue = previousProject == null ? null : previousProject.customs().get(customMetadataType);
175+
Object value = project == null ? null : project.customs().get(customMetadataType);
176+
return Objects.equals(previousValue, value) == false;
183177
}
184178

185179
private <C extends Metadata.MetadataCustom<C>> Set<String> changedCustoms(

0 commit comments

Comments
 (0)