You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2022/10/12 12:20:56 UTC
[GitHub] [ignite-3] vldpyatkov opened a new pull request, #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
vldpyatkov opened a new pull request, #1197:
URL: https://github.com/apache/ignite-3/pull/1197
…n logic for RO reads
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite-3] sergeyuttsel commented on a diff in pull request #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
Posted by GitBox <gi...@apache.org>.
sergeyuttsel commented on code in PR #1197:
URL: https://github.com/apache/ignite-3/pull/1197#discussion_r998413648
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -421,6 +542,38 @@ private CompletableFuture<Object> processTxFinishAction(TxFinishReplicaRequest r
return allOf(cleanupFutures).thenApply(ignored -> null);
}
+ /**
+ * Finishes a transaction.
+ *
+ * @param aggregatedGroupIds Replication groups identifies which are enlisted in the transaction.
+ * @param txId Transaction id.
+ * @param commit True is the transaction is committed, false otherwise.
+ * @return Future to wait of the finish.
+ */
+ private CompletableFuture<Object> finishTransaction(List<String> aggregatedGroupIds, UUID txId, boolean commit) {
+ // TODO: Timestamp from request is not using until IGNITE-XXXX (request.commitTimestamp())
Review Comment:
ticket number
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite-3] vldpyatkov commented on a diff in pull request #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
Posted by GitBox <gi...@apache.org>.
vldpyatkov commented on code in PR #1197:
URL: https://github.com/apache/ignite-3/pull/1197#discussion_r995720048
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PlacementDriver.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed.replicator;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.tx.TxMeta;
+import org.apache.ignite.internal.tx.message.TxStateReplicaRequest;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Placement driver.
+ */
+public class PlacementDriver {
+ /** Assignment nodes per replication group. */
+ private final Map<String, LinkedHashSet<ClusterNode>> primaryReplicaMapping = new ConcurrentHashMap<>();
+
+ /** Replication service. */
+ private final ReplicaService replicaService;
+
+ /**
+ * The constructor.
+ *
+ * @param replicaService Replication service.
+ */
+ public PlacementDriver(ReplicaService replicaService) {
+ this.replicaService = replicaService;
+ }
+
+ /**
+ * Sent a transaction sate request to the primary replica.
+ *
+ * @param replicaGrp Replication group id.
+ * @param request Status request.
+ * @return Result future.
+ */
+ public CompletableFuture<TxMeta> sentMetaRequest(String replicaGrp, TxStateReplicaRequest request) {
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite-3] sergeyuttsel commented on a diff in pull request #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
Posted by GitBox <gi...@apache.org>.
sergeyuttsel commented on code in PR #1197:
URL: https://github.com/apache/ignite-3/pull/1197#discussion_r995660065
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -220,6 +221,9 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
/** Data storage manager. */
private final DataStorageManager dataStorageMgr;
+ /** Placement Driver . */
Review Comment:
Remove extra space before dot.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -200,6 +247,66 @@ public CompletableFuture<Object> invoke(ReplicaRequest request) {
});
}
+ /**
+ * Processes a transaction state request.
+ *
+ * @param request Transaction state request.
+ * @return Result future.
+ */
+ private CompletableFuture<Object> processTxStateReplicaRequest(TxStateReplicaRequest request) {
+ TxStateReplicaRequest txStateReq = request;
+
+ return raftClient.refreshAndGetLeaderWithTerm()
+ .thenCompose(replicaAndTerm -> {
+ NetworkAddress leaderAddress = replicaAndTerm.get1().address();
+
+ if (topologyService.localMember().address().equals(leaderAddress)) {
+
+ TxMeta tsState = getTxStateConcurrently(txStateReq);
+
+ return CompletableFuture.completedFuture(new IgniteBiTuple<>(tsState, null));
+ } else {
+ return CompletableFuture.completedFuture(
+ new IgniteBiTuple<>(null, topologyService.getByAddress(leaderAddress)));
+ }
+ }
+ );
+ }
+
+ /**
+ * Gets a transaction state or {@code null}, if the transaction is not completed.
+ *
+ * @param txStateReq Transaction state request.
+ * @return Transaction state meta of {@code null}.
Review Comment:
Maybe "or null"
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1169,9 +1351,16 @@ private CompletableFuture<Void> ensureReplicaIsPrimary(ReplicaRequest request) {
*
* @param readResult Read result to resolve.
* @param txId Nullable transaction id, should be provided if resolution is performed within the context of RW transaction.
+ * @param timestamp Timestamp is used is RO transaction only.
+ * @param lastCommitted Action to get the latest committed row, it is used is RO transaction only.
Review Comment:
"in RO"
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PlacementDriver.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed.replicator;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.tx.TxMeta;
+import org.apache.ignite.internal.tx.message.TxStateReplicaRequest;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Placement driver.
+ */
+public class PlacementDriver {
+ /** Assignment nodes per replication group. */
+ private final Map<String, LinkedHashSet<ClusterNode>> primaryReplicaMapping = new ConcurrentHashMap<>();
+
+ /** Replication service. */
+ private final ReplicaService replicaService;
+
+ /**
+ * The constructor.
+ *
+ * @param replicaService Replication service.
+ */
+ public PlacementDriver(ReplicaService replicaService) {
+ this.replicaService = replicaService;
+ }
+
+ /**
+ * Sent a transaction sate request to the primary replica.
+ *
+ * @param replicaGrp Replication group id.
+ * @param request Status request.
+ * @return Result future.
+ */
+ public CompletableFuture<TxMeta> sentMetaRequest(String replicaGrp, TxStateReplicaRequest request) {
+ CompletableFuture<TxMeta> resFut = new CompletableFuture<>();
+
+ sentAndRetry(resFut, replicaGrp, request);
+
+ return resFut;
+ }
+
+ /**
+ * Updates an assignment for the specific replication group.
+ *
+ * @param replicaGrpId Replication group id.
+ * @param assignment Assignment.
+ */
+ public void updateAssignment(String replicaGrpId, Collection<ClusterNode> assignment) {
+ primaryReplicaMapping.put(replicaGrpId, new LinkedHashSet<>(assignment));
+ }
+
+ /**
+ * Sent a request to primary replica of the replication group.
+ * If the primary node is changed the request is sending to a new primary node.
+ *
+ * @param resFut Response future.
+ * @param replicaGrp Replication group id.
+ * @param request Request.
+ */
+ private void sentAndRetry(CompletableFuture<TxMeta> resFut, String replicaGrp, TxStateReplicaRequest request) {
+ ClusterNode nodeToSent = primaryReplicaMapping.get(replicaGrp).iterator().next();
+
+ //TODO: Handle timout here. If an invocation is timed out the request should be passed to a next node from the assignment.
Review Comment:
Need to create ticket.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1169,9 +1351,16 @@ private CompletableFuture<Void> ensureReplicaIsPrimary(ReplicaRequest request) {
*
* @param readResult Read result to resolve.
* @param txId Nullable transaction id, should be provided if resolution is performed within the context of RW transaction.
+ * @param timestamp Timestamp is used is RO transaction only.
Review Comment:
"in RO"
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -200,6 +247,66 @@ public CompletableFuture<Object> invoke(ReplicaRequest request) {
});
}
+ /**
+ * Processes a transaction state request.
+ *
+ * @param request Transaction state request.
+ * @return Result future.
+ */
+ private CompletableFuture<Object> processTxStateReplicaRequest(TxStateReplicaRequest request) {
+ TxStateReplicaRequest txStateReq = request;
+
+ return raftClient.refreshAndGetLeaderWithTerm()
+ .thenCompose(replicaAndTerm -> {
+ NetworkAddress leaderAddress = replicaAndTerm.get1().address();
+
+ if (topologyService.localMember().address().equals(leaderAddress)) {
+
+ TxMeta tsState = getTxStateConcurrently(txStateReq);
+
+ return CompletableFuture.completedFuture(new IgniteBiTuple<>(tsState, null));
+ } else {
+ return CompletableFuture.completedFuture(
+ new IgniteBiTuple<>(null, topologyService.getByAddress(leaderAddress)));
+ }
+ }
+ );
+ }
+
+ /**
+ * Gets a transaction state or {@code null}, if the transaction is not completed.
+ *
+ * @param txStateReq Transaction state request.
+ * @return Transaction state meta of {@code null}.
+ */
+ private TxMeta getTxStateConcurrently(TxStateReplicaRequest txStateReq) {
+ //TODO: IGNITE-XXXX review this after the commit timestamp will be provided from a commit request (request.commitTimestamp()).
Review Comment:
ticket number
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -117,16 +130,39 @@ public class PartitionReplicaListener implements ReplicaListener {
*/
private final ConcurrentNavigableMap<IgniteUuid, PartitionTimestampCursor> cursors;
+ /** Tx state storage. */
+ private final TxStateStorage txStateStorage;
+
+ /** Topology service. */
+ private final TopologyService topologyService;
+
+ /** Hybrid clock. */
+ private final HybridClock hybridClock;
+
+ /** Placement Driver. */
+ private final PlacementDriver placementDriver;
+
+ /**
+ * Map to control clock's update in the read only transactions concurrently with a commit timestamp.
+ * TODO: IGNITE-XXXX review this after the commit timestamp will be provided from a commit request (request.commitTimestamp()).
Review Comment:
Need to specify the number of an existing ticket or create a new one.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PlacementDriver.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed.replicator;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.tx.TxMeta;
+import org.apache.ignite.internal.tx.message.TxStateReplicaRequest;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Placement driver.
+ */
+public class PlacementDriver {
+ /** Assignment nodes per replication group. */
+ private final Map<String, LinkedHashSet<ClusterNode>> primaryReplicaMapping = new ConcurrentHashMap<>();
+
+ /** Replication service. */
+ private final ReplicaService replicaService;
+
+ /**
+ * The constructor.
+ *
+ * @param replicaService Replication service.
+ */
+ public PlacementDriver(ReplicaService replicaService) {
+ this.replicaService = replicaService;
+ }
+
+ /**
+ * Sent a transaction sate request to the primary replica.
+ *
+ * @param replicaGrp Replication group id.
+ * @param request Status request.
+ * @return Result future.
+ */
+ public CompletableFuture<TxMeta> sentMetaRequest(String replicaGrp, TxStateReplicaRequest request) {
+ CompletableFuture<TxMeta> resFut = new CompletableFuture<>();
+
+ sentAndRetry(resFut, replicaGrp, request);
+
+ return resFut;
+ }
+
+ /**
+ * Updates an assignment for the specific replication group.
+ *
+ * @param replicaGrpId Replication group id.
+ * @param assignment Assignment.
+ */
+ public void updateAssignment(String replicaGrpId, Collection<ClusterNode> assignment) {
+ primaryReplicaMapping.put(replicaGrpId, new LinkedHashSet<>(assignment));
+ }
+
+ /**
+ * Sent a request to primary replica of the replication group.
Review Comment:
the first replica, not a primary.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PlacementDriver.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed.replicator;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.tx.TxMeta;
+import org.apache.ignite.internal.tx.message.TxStateReplicaRequest;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Placement driver.
+ */
+public class PlacementDriver {
+ /** Assignment nodes per replication group. */
+ private final Map<String, LinkedHashSet<ClusterNode>> primaryReplicaMapping = new ConcurrentHashMap<>();
+
+ /** Replication service. */
+ private final ReplicaService replicaService;
+
+ /**
+ * The constructor.
+ *
+ * @param replicaService Replication service.
+ */
+ public PlacementDriver(ReplicaService replicaService) {
+ this.replicaService = replicaService;
+ }
+
+ /**
+ * Sent a transaction sate request to the primary replica.
+ *
+ * @param replicaGrp Replication group id.
+ * @param request Status request.
+ * @return Result future.
+ */
+ public CompletableFuture<TxMeta> sentMetaRequest(String replicaGrp, TxStateReplicaRequest request) {
Review Comment:
"send" here and below.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite-3] sergeyuttsel commented on a diff in pull request #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
Posted by GitBox <gi...@apache.org>.
sergeyuttsel commented on code in PR #1197:
URL: https://github.com/apache/ignite-3/pull/1197#discussion_r998415320
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PlacementDriver.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed.replicator;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.tx.TxMeta;
+import org.apache.ignite.internal.tx.message.TxStateReplicaRequest;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Placement driver.
+ */
+public class PlacementDriver {
+ /** Assignment nodes per replication group. */
+ private final Map<String, LinkedHashSet<ClusterNode>> primaryReplicaMapping = new ConcurrentHashMap<>();
+
+ /** Replication service. */
+ private final ReplicaService replicaService;
+
+ /**
+ * The constructor.
+ *
+ * @param replicaService Replication service.
+ */
+ public PlacementDriver(ReplicaService replicaService) {
+ this.replicaService = replicaService;
+ }
+
+ /**
+ * Sent a transaction sate request to the primary replica.
+ *
+ * @param replicaGrp Replication group id.
+ * @param request Status request.
+ * @return Result future.
+ */
+ public CompletableFuture<TxMeta> sentMetaRequest(String replicaGrp, TxStateReplicaRequest request) {
+ CompletableFuture<TxMeta> resFut = new CompletableFuture<>();
+
+ sentAndRetry(resFut, replicaGrp, request);
+
+ return resFut;
+ }
+
+ /**
+ * Updates an assignment for the specific replication group.
+ *
+ * @param replicaGrpId Replication group id.
+ * @param assignment Assignment.
+ */
+ public void updateAssignment(String replicaGrpId, Collection<ClusterNode> assignment) {
+ primaryReplicaMapping.put(replicaGrpId, new LinkedHashSet<>(assignment));
+ }
+
+ /**
+ * Sent a request to primary replica of the replication group.
Review Comment:
That's better
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite-3] vldpyatkov commented on a diff in pull request #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
Posted by GitBox <gi...@apache.org>.
vldpyatkov commented on code in PR #1197:
URL: https://github.com/apache/ignite-3/pull/1197#discussion_r995727105
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PlacementDriver.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed.replicator;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.tx.TxMeta;
+import org.apache.ignite.internal.tx.message.TxStateReplicaRequest;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Placement driver.
+ */
+public class PlacementDriver {
+ /** Assignment nodes per replication group. */
+ private final Map<String, LinkedHashSet<ClusterNode>> primaryReplicaMapping = new ConcurrentHashMap<>();
+
+ /** Replication service. */
+ private final ReplicaService replicaService;
+
+ /**
+ * The constructor.
+ *
+ * @param replicaService Replication service.
+ */
+ public PlacementDriver(ReplicaService replicaService) {
+ this.replicaService = replicaService;
+ }
+
+ /**
+ * Sent a transaction sate request to the primary replica.
+ *
+ * @param replicaGrp Replication group id.
+ * @param request Status request.
+ * @return Result future.
+ */
+ public CompletableFuture<TxMeta> sentMetaRequest(String replicaGrp, TxStateReplicaRequest request) {
+ CompletableFuture<TxMeta> resFut = new CompletableFuture<>();
+
+ sentAndRetry(resFut, replicaGrp, request);
+
+ return resFut;
+ }
+
+ /**
+ * Updates an assignment for the specific replication group.
+ *
+ * @param replicaGrpId Replication group id.
+ * @param assignment Assignment.
+ */
+ public void updateAssignment(String replicaGrpId, Collection<ClusterNode> assignment) {
+ primaryReplicaMapping.put(replicaGrpId, new LinkedHashSet<>(assignment));
+ }
+
+ /**
+ * Sent a request to primary replica of the replication group.
Review Comment:
It tries to send it on primary node.
I tried to explain those concept in another terms, please look it again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite-3] vldpyatkov commented on a diff in pull request #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
Posted by GitBox <gi...@apache.org>.
vldpyatkov commented on code in PR #1197:
URL: https://github.com/apache/ignite-3/pull/1197#discussion_r995718849
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1169,9 +1351,16 @@ private CompletableFuture<Void> ensureReplicaIsPrimary(ReplicaRequest request) {
*
* @param readResult Read result to resolve.
* @param txId Nullable transaction id, should be provided if resolution is performed within the context of RW transaction.
+ * @param timestamp Timestamp is used is RO transaction only.
Review Comment:
Done.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1169,9 +1351,16 @@ private CompletableFuture<Void> ensureReplicaIsPrimary(ReplicaRequest request) {
*
* @param readResult Read result to resolve.
* @param txId Nullable transaction id, should be provided if resolution is performed within the context of RW transaction.
+ * @param timestamp Timestamp is used is RO transaction only.
+ * @param lastCommitted Action to get the latest committed row, it is used is RO transaction only.
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite-3] vldpyatkov merged pull request #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
Posted by GitBox <gi...@apache.org>.
vldpyatkov merged PR #1197:
URL: https://github.com/apache/ignite-3/pull/1197
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite-3] vldpyatkov commented on a diff in pull request #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
Posted by GitBox <gi...@apache.org>.
vldpyatkov commented on code in PR #1197:
URL: https://github.com/apache/ignite-3/pull/1197#discussion_r995717599
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -200,6 +247,66 @@ public CompletableFuture<Object> invoke(ReplicaRequest request) {
});
}
+ /**
+ * Processes a transaction state request.
+ *
+ * @param request Transaction state request.
+ * @return Result future.
+ */
+ private CompletableFuture<Object> processTxStateReplicaRequest(TxStateReplicaRequest request) {
+ TxStateReplicaRequest txStateReq = request;
+
+ return raftClient.refreshAndGetLeaderWithTerm()
+ .thenCompose(replicaAndTerm -> {
+ NetworkAddress leaderAddress = replicaAndTerm.get1().address();
+
+ if (topologyService.localMember().address().equals(leaderAddress)) {
+
+ TxMeta tsState = getTxStateConcurrently(txStateReq);
+
+ return CompletableFuture.completedFuture(new IgniteBiTuple<>(tsState, null));
+ } else {
+ return CompletableFuture.completedFuture(
+ new IgniteBiTuple<>(null, topologyService.getByAddress(leaderAddress)));
+ }
+ }
+ );
+ }
+
+ /**
+ * Gets a transaction state or {@code null}, if the transaction is not completed.
+ *
+ * @param txStateReq Transaction state request.
+ * @return Transaction state meta of {@code null}.
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite-3] vldpyatkov commented on a diff in pull request #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
Posted by GitBox <gi...@apache.org>.
vldpyatkov commented on code in PR #1197:
URL: https://github.com/apache/ignite-3/pull/1197#discussion_r995717192
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -220,6 +221,9 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
/** Data storage manager. */
private final DataStorageManager dataStorageMgr;
+ /** Placement Driver . */
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite-3] vldpyatkov commented on a diff in pull request #1197: IGNITE-17637 Implement a commit partition path write intent resolutio…
Posted by GitBox <gi...@apache.org>.
vldpyatkov commented on code in PR #1197:
URL: https://github.com/apache/ignite-3/pull/1197#discussion_r996781961
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PlacementDriver.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed.replicator;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.tx.TxMeta;
+import org.apache.ignite.internal.tx.message.TxStateReplicaRequest;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Placement driver.
+ */
+public class PlacementDriver {
+ /** Assignment nodes per replication group. */
+ private final Map<String, LinkedHashSet<ClusterNode>> primaryReplicaMapping = new ConcurrentHashMap<>();
+
+ /** Replication service. */
+ private final ReplicaService replicaService;
+
+ /**
+ * The constructor.
+ *
+ * @param replicaService Replication service.
+ */
+ public PlacementDriver(ReplicaService replicaService) {
+ this.replicaService = replicaService;
+ }
+
+ /**
+ * Sent a transaction sate request to the primary replica.
+ *
+ * @param replicaGrp Replication group id.
+ * @param request Status request.
+ * @return Result future.
+ */
+ public CompletableFuture<TxMeta> sentMetaRequest(String replicaGrp, TxStateReplicaRequest request) {
+ CompletableFuture<TxMeta> resFut = new CompletableFuture<>();
+
+ sentAndRetry(resFut, replicaGrp, request);
+
+ return resFut;
+ }
+
+ /**
+ * Updates an assignment for the specific replication group.
+ *
+ * @param replicaGrpId Replication group id.
+ * @param assignment Assignment.
+ */
+ public void updateAssignment(String replicaGrpId, Collection<ClusterNode> assignment) {
+ primaryReplicaMapping.put(replicaGrpId, new LinkedHashSet<>(assignment));
+ }
+
+ /**
+ * Sent a request to primary replica of the replication group.
+ * If the primary node is changed the request is sending to a new primary node.
+ *
+ * @param resFut Response future.
+ * @param replicaGrp Replication group id.
+ * @param request Request.
+ */
+ private void sentAndRetry(CompletableFuture<TxMeta> resFut, String replicaGrp, TxStateReplicaRequest request) {
+ ClusterNode nodeToSent = primaryReplicaMapping.get(replicaGrp).iterator().next();
+
+ //TODO: Handle timout here. If an invocation is timed out the request should be passed to a next node from the assignment.
Review Comment:
I removed the comment, because our transaction implementation is designed for stable topology only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org