Perform resource existence checks only once in Mongo connector [HZ-2673] #24953
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,195 @@ | ||
| /* | ||
| * Copyright 2023 Hazelcast Inc. | ||
| * | ||
| * Licensed under the Hazelcast Community License (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://hazelcast.com/hazelcast-community-license | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.hazelcast.jet.mongodb.impl; | ||
|
|
||
| import com.hazelcast.cluster.Address; | ||
| import com.hazelcast.dataconnection.DataConnection; | ||
| import com.hazelcast.dataconnection.impl.InternalDataConnectionService; | ||
| import com.hazelcast.function.SupplierEx; | ||
| import com.hazelcast.jet.JetException; | ||
| import com.hazelcast.jet.core.ProcessorMetaSupplier; | ||
| import com.hazelcast.jet.core.ProcessorSupplier; | ||
| import com.hazelcast.jet.core.Vertex; | ||
| import com.hazelcast.jet.datamodel.Tuple2; | ||
| import com.hazelcast.jet.impl.processor.ExpectNothingP; | ||
| import com.hazelcast.jet.impl.util.Util; | ||
| import com.hazelcast.jet.mongodb.dataconnection.MongoDataConnection; | ||
| import com.hazelcast.jet.pipeline.DataConnectionRef; | ||
| import com.hazelcast.spi.impl.NodeEngineImpl; | ||
| import com.mongodb.client.MongoClient; | ||
| import com.mongodb.client.MongoDatabase; | ||
|
|
||
| import javax.annotation.Nonnull; | ||
| import javax.annotation.Nullable; | ||
| import java.security.Permission; | ||
| import java.util.List; | ||
| import java.util.Map.Entry; | ||
| import java.util.function.Function; | ||
|
|
||
| import static com.hazelcast.internal.util.UuidUtil.newUnsecureUuidString; | ||
| import static com.hazelcast.jet.datamodel.Tuple2.tuple2; | ||
| import static com.hazelcast.jet.impl.util.Util.arrayIndexOf; | ||
| import static com.hazelcast.jet.mongodb.impl.MongoUtilities.checkCollectionExists; | ||
| import static com.hazelcast.jet.mongodb.impl.MongoUtilities.checkDatabaseExists; | ||
| import static com.hazelcast.partition.strategy.StringPartitioningStrategy.getPartitionKey; | ||
| import static java.util.Collections.singletonList; | ||
|
|
||
| /** | ||
| * A {@link ProcessorMetaSupplier} that will check if requested database and collection exist before creating | ||
| * the processors. | ||
| */ | ||
| public class DbCheckingPMetaSupplier implements ProcessorMetaSupplier { | ||
|
|
||
| private final Permission requiredPermission; | ||
| private final boolean shouldCheck; | ||
| private boolean forceTotalParallelismOne; | ||
| private final String databaseName; | ||
| private final String collectionName; | ||
| private final ProcessorSupplier processorSupplier; | ||
| private final SupplierEx<? extends MongoClient> clientSupplier; | ||
| private final DataConnectionRef dataConnectionRef; | ||
| private int preferredLocalParallelism = Vertex.LOCAL_PARALLELISM_USE_DEFAULT; | ||
|
|
||
| private transient Address ownerAddress; | ||
k-jamroz marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| /** | ||
| * Creates a new instance of this meta supplier. | ||
| */ | ||
| public DbCheckingPMetaSupplier(@Nullable Permission requiredPermission, | ||
| boolean shouldCheck, | ||
| boolean forceTotalParallelismOne, | ||
| @Nullable String databaseName, | ||
| @Nullable String collectionName, | ||
| @Nullable SupplierEx<? extends MongoClient> clientSupplier, | ||
| @Nullable DataConnectionRef dataConnectionRef, | ||
| @Nonnull ProcessorSupplier processorSupplier) { | ||
| this.requiredPermission = requiredPermission; | ||
| this.shouldCheck = shouldCheck; | ||
| this.forceTotalParallelismOne = forceTotalParallelismOne; | ||
| this.databaseName = databaseName; | ||
| this.collectionName = collectionName; | ||
| this.processorSupplier = processorSupplier; | ||
| this.clientSupplier = clientSupplier; | ||
| this.dataConnectionRef = dataConnectionRef; | ||
| } | ||
|
|
||
| /** | ||
| * Sets preferred local parallelism. If {@link #forceTotalParallelismOne} is selected, this | ||
| * method will have no effect. | ||
| */ | ||
| public DbCheckingPMetaSupplier withPreferredLocalParallelism(int preferredLocalParallelism) { | ||
k-jamroz marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| this.preferredLocalParallelism = forceTotalParallelismOne ? 1 : preferredLocalParallelism; | ||
k-jamroz marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public int preferredLocalParallelism() { | ||
k-jamroz marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| return forceTotalParallelismOne ? 1 : preferredLocalParallelism; | ||
| } | ||
|
|
||
| @Nullable | ||
| @Override | ||
| public Permission getRequiredPermission() { | ||
| return requiredPermission; | ||
| } | ||
|
|
||
| /** | ||
| * If true, only one instance of given supplier will be created. | ||
| */ | ||
| public DbCheckingPMetaSupplier forceTotalParallelismOne(boolean forceTotalParallelismOne) { | ||
| this.forceTotalParallelismOne = forceTotalParallelismOne; | ||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean initIsCooperative() { | ||
| return true; | ||
k-jamroz marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } | ||
|
|
||
| @Override | ||
| public void init(@Nonnull Context context) throws Exception { | ||
| if (forceTotalParallelismOne) { | ||
| preferredLocalParallelism = 1; | ||
k-jamroz marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| if (context.localParallelism() != 1) { | ||
| throw new IllegalArgumentException( | ||
| "Local parallelism of " + context.localParallelism() + " was requested for a vertex that " | ||
| + "supports only total parallelism of 1. Local parallelism must be 1."); | ||
| } | ||
| String key = getPartitionKey(newUnsecureUuidString()); | ||
| int partitionId = context.hazelcastInstance().getPartitionService().getPartition(key).getPartitionId(); | ||
k-jamroz marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| ownerAddress = context.partitionAssignment().entrySet().stream() | ||
| .filter(en -> arrayIndexOf(partitionId, en.getValue()) >= 0) | ||
| .findAny() | ||
| .map(Entry::getKey) | ||
| .orElseThrow(() -> new RuntimeException("Owner partition not assigned to any " + | ||
| "participating member")); | ||
| } | ||
|
|
||
| if (shouldCheck) { | ||
| Tuple2<MongoClient, DataConnection> clientAndRef = connect(context); | ||
| MongoClient client = clientAndRef.requiredF0(); | ||
|
There was a problem hiding this comment. Where is this client instance closed? There was a problem hiding this comment. Good catch, 726d39d |
||
| try { | ||
| if (databaseName != null) { | ||
| checkDatabaseExists(client, databaseName); | ||
| MongoDatabase database = client.getDatabase(databaseName); | ||
| if (collectionName != null) { | ||
| checkCollectionExists(database, collectionName); | ||
| } | ||
| } | ||
| } finally { | ||
| DataConnection connection = clientAndRef.f1(); | ||
| if (connection != null) { | ||
| connection.release(); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private Tuple2<MongoClient, DataConnection> connect(Context context) { | ||
| try { | ||
| if (clientSupplier != null) { | ||
| return tuple2(clientSupplier.get(), null); | ||
| } else if (dataConnectionRef != null) { | ||
| NodeEngineImpl nodeEngine = Util.getNodeEngine(context.hazelcastInstance()); | ||
| InternalDataConnectionService dataConnectionService = nodeEngine.getDataConnectionService(); | ||
k-jamroz marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| var dataConnection = dataConnectionService.getAndRetainDataConnection(dataConnectionRef.getName(), | ||
| MongoDataConnection.class); | ||
| return tuple2(dataConnection.getClient(), dataConnection); | ||
| } else { | ||
| throw new IllegalArgumentException("Either connectionSupplier or dataConnectionRef must be provided " + | ||
| "if database and collection existence checks are requested"); | ||
| } | ||
| } catch (Exception e) { | ||
| throw new JetException("Cannot connect to MongoDB", e); | ||
| } | ||
| } | ||
|
|
||
| @Nonnull | ||
| @Override | ||
| public Function<? super Address, ? extends ProcessorSupplier> get(@Nonnull List<Address> addresses) { | ||
| if (forceTotalParallelismOne) { | ||
| return addr -> addr.equals(ownerAddress) ? processorSupplier : count -> singletonList(new ExpectNothingP()); | ||
| } else { | ||
| return addr -> processorSupplier; | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean closeIsCooperative() { | ||
| return true; | ||
| } | ||
|
|
||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| /* | ||
| * Copyright 2023 Hazelcast Inc. | ||
| * | ||
| * Licensed under the Hazelcast Community License (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://hazelcast.com/hazelcast-community-license | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.hazelcast.jet.mongodb.impl; | ||
|
|
||
| import com.hazelcast.function.SupplierEx; | ||
| import com.hazelcast.jet.core.ProcessorSupplier; | ||
| import com.hazelcast.jet.pipeline.DataConnectionRef; | ||
| import com.mongodb.client.MongoClient; | ||
|
|
||
| import java.security.Permission; | ||
|
|
||
| public class DbCheckingPMetaSupplierBuilder { | ||
| private Permission requiredPermission; | ||
| private boolean shouldCheck; | ||
| private boolean forceTotalParallelismOne; | ||
| private String databaseName; | ||
| private String collectionName; | ||
| private SupplierEx<? extends MongoClient> clientSupplier; | ||
| private DataConnectionRef dataConnectionRef; | ||
| private ProcessorSupplier processorSupplier; | ||
|
|
||
| public DbCheckingPMetaSupplierBuilder setRequiredPermission(Permission requiredPermission) { | ||
| this.requiredPermission = requiredPermission; | ||
| return this; | ||
| } | ||
|
|
||
| public DbCheckingPMetaSupplierBuilder setShouldCheck(boolean shouldCheck) { | ||
| this.shouldCheck = shouldCheck; | ||
| return this; | ||
| } | ||
|
|
||
| public DbCheckingPMetaSupplierBuilder setForceTotalParallelismOne(boolean forceTotalParallelismOne) { | ||
| this.forceTotalParallelismOne = forceTotalParallelismOne; | ||
| return this; | ||
| } | ||
|
|
||
| public DbCheckingPMetaSupplierBuilder setDatabaseName(String databaseName) { | ||
k-jamroz marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| this.databaseName = databaseName; | ||
| return this; | ||
| } | ||
|
|
||
| public DbCheckingPMetaSupplierBuilder setCollectionName(String collectionName) { | ||
| this.collectionName = collectionName; | ||
| return this; | ||
| } | ||
|
|
||
| public DbCheckingPMetaSupplierBuilder setClientSupplier(SupplierEx<? extends MongoClient> clientSupplier) { | ||
| this.clientSupplier = clientSupplier; | ||
| return this; | ||
| } | ||
|
|
||
| public DbCheckingPMetaSupplierBuilder setDataConnectionRef(DataConnectionRef dataConnectionRef) { | ||
| this.dataConnectionRef = dataConnectionRef; | ||
| return this; | ||
| } | ||
|
|
||
| public DbCheckingPMetaSupplierBuilder setProcessorSupplier(ProcessorSupplier processorSupplier) { | ||
| this.processorSupplier = processorSupplier; | ||
| return this; | ||
| } | ||
|
|
||
| public DbCheckingPMetaSupplier create() { | ||
frant-hartm marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| return new DbCheckingPMetaSupplier(requiredPermission, shouldCheck, forceTotalParallelismOne, databaseName, | ||
| collectionName, clientSupplier, dataConnectionRef, processorSupplier); | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
what is the point of having
localParamsvariable?ReadMongoParamsis mutable andReadMongoParams.setXXXmethods do not return new instance.I can see that it is used in lambda in
withProcessorSupplierbut all sources created from the same builder will share the same instance ofparams. If the builder is modified afterbuild()you can have problems (local variable fixes compilation problem but not the sharing problem)There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
treating builder as a prototype might not be a best idea but someone might try do that
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
usage of
localParamsthroughout this methods makes me think that maybe you wanted to clone themThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
we have similar situation in
MongoSinkBuilderThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Referencing fields of the builder inside lambda will catch the reference to the builder, meaning that whole builder would need to become serializable. Referencing local variable instead solves the issue.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yes, but it does not solve problem with builder as prototype. We do not control lifecycle of the builder and the time when PMS will be serialized.