You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "sanpwc (via GitHub)" <gi...@apache.org> on 2023/09/28 16:02:47 UTC
[GitHub] [ignite-3] sanpwc commented on a diff in pull request #2520: IGNITE-20127
sanpwc commented on code in PR #2520:
URL: https://github.com/apache/ignite-3/pull/2520#discussion_r1340203900
##########
modules/network-api/src/main/java/org/apache/ignite/network/MessagingService.java:
##########
@@ -42,6 +42,7 @@ public interface MessagingService {
* <li>Messages send to same receiver will be delivered in the same order as they were sent;</li>
* <li>If a message N has been successfully delivered to a member implies that all messages to same receiver
* preceding N have also been successfully delivered.</li>
+ * <li>A message is delivered only once</li>
Review Comment:
Full stop is missing in the end.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -261,6 +261,35 @@ private void onReplicaMessageReceived(NetworkMessage message, String senderConsi
}
}
+ if (res.repFuture() != null) {
+ assert request instanceof PrimaryReplicaRequest;
+
+ res.repFuture().handle((res0, ex0) -> {
+ NetworkMessage msg0;
+
+ LOG.debug("Sending delayed response for replica request [request={}]", request);
+
+ if (ex == null) {
+ msg0 = prepareReplicaResponse(finalSendTimestamp, res0);
Review Comment:
What kind of result you expect in repFut (res0). I believe that it is void in any case because it basically tells us that replication is finished or throws an exception.
##########
modules/network/src/main/java/org/apache/ignite/network/DefaultMessagingService.java:
##########
@@ -115,6 +115,8 @@ public DefaultMessagingService(
this.marshaller = marshaller;
this.outboundExecutor = Executors.newSingleThreadExecutor(NamedThreadFactory.create(nodeName, "MessagingService-outbound-", LOG));
+ // TODO asch the implementation of delayed acks relies on absence of reordering on subsequent messages delivery.
Review Comment:
Why?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -261,6 +261,35 @@ private void onReplicaMessageReceived(NetworkMessage message, String senderConsi
}
}
+ if (res.repFuture() != null) {
+ assert request instanceof PrimaryReplicaRequest;
+
+ res.repFuture().handle((res0, ex0) -> {
+ NetworkMessage msg0;
+
+ LOG.debug("Sending delayed response for replica request [request={}]", request);
+
+ if (ex == null) {
+ msg0 = prepareReplicaResponse(finalSendTimestamp, res0);
+ } else {
+ LOG.warn("Failed to process delayed response [request={}]", ex, request);
+
+ msg0 = prepareReplicaErrorResponse(finalSendTimestamp, ex);
Review Comment:
I don't think that it's valid to use same finalSendTimestamp with both fromPrimary and delayed responses.
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java:
##########
@@ -176,6 +177,7 @@ public <RowT> CompletableFuture<?> upsertAll(
.term(nodeWithTerm.term())
.requestType(RequestType.RW_UPSERT_ALL)
.timestampLong(clock.nowLong())
+ .skipDelayedAck(true)
Review Comment:
What does it mean and why do we skip delayed ack for _ALL operations?
##########
modules/table/src/testFixtures/java/org/apache/ignite/internal/table/TxAbstractTest.java:
##########
@@ -1541,7 +1580,7 @@ public void run() {
}
while (!stop.get() && firstErr.get() == null) {
- InternalTransaction tx = txManager(accounts).begin();
+ InternalTransaction tx = clientTxManager().begin();
Review Comment:
Is an intended change? I mean that, despite the fact that we still have txManager() we want use client ones, right?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -326,12 +331,18 @@ public PartitionReplicaListener(
}
@Override
- public CompletableFuture<?> invoke(ReplicaRequest request) {
+ public CompletableFuture<ReplicaResult> invoke(ReplicaRequest request) {
if (request instanceof TxStateReplicaRequest) {
- return processTxStateReplicaRequest((TxStateReplicaRequest) request);
+ return processTxStateReplicaRequest((TxStateReplicaRequest) request).thenApply(res -> new ReplicaResult(res, null));
}
- return ensureReplicaIsPrimary(request).thenCompose(isPrimary -> processRequest(request, isPrimary));
+ return ensureReplicaIsPrimary(request).thenCompose(isPrimary -> processRequest(request, isPrimary).thenApply(res -> {
+ if (res instanceof ReplicaResult) {
+ return (ReplicaResult) res;
+ } else {
+ return new ReplicaResult(res, null);
Review Comment:
I'd rather introduce overloaded version with one param.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -261,6 +261,35 @@ private void onReplicaMessageReceived(NetworkMessage message, String senderConsi
}
}
+ if (res.repFuture() != null) {
Review Comment:
How do we handle single noded raft groups? I didn't check the code below, maybe it's already implemented. But for now, my questions/expectations are following:
- In case of single noded raft group and if primary is collocated with that single node, we should skip primary storage updates, and just send the command for the replication, synchronously await the result and return the future with completed res.repFuture and thus send single response to the client side.
- Otherwise use common logic with two responses.
- In all cases we should check whether repFuture is completed when we are ready to send response from the primary and send only one response if it is.
BTW if items 1 and 2 aren't implemented yet, I'd rather do it in a separate ticket.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -261,6 +261,35 @@ private void onReplicaMessageReceived(NetworkMessage message, String senderConsi
}
}
+ if (res.repFuture() != null) {
+ assert request instanceof PrimaryReplicaRequest;
+
+ res.repFuture().handle((res0, ex0) -> {
+ NetworkMessage msg0;
+
+ LOG.debug("Sending delayed response for replica request [request={}]", request);
+
+ if (ex == null) {
+ msg0 = prepareReplicaResponse(finalSendTimestamp, res0);
+ } else {
+ LOG.warn("Failed to process delayed response [request={}]", ex, request);
+
+ msg0 = prepareReplicaErrorResponse(finalSendTimestamp, ex);
+ }
+
+ ClusterNode sender = clusterNetSvc.topologyService().getByConsistentId(senderConsistentId);
Review Comment:
I'd rather add overloaded version of send with recipientConsistentId as we discussed with Ivan and just skip the clusterNodeResolver part.
We already have one with channelType param
`CompletableFuture<Void> send(String recipientConsistentId, ChannelType channelType, NetworkMessage msg);`
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java:
##########
@@ -237,14 +239,15 @@ public <RowT> CompletableFuture<?> insertAll(
TablePartitionId partGroupId = new TablePartitionId(tableId, partToRows.getIntKey());
NodeWithTerm nodeWithTerm = ectx.description().mapping().updatingTableAssignments().get(partToRows.getIntKey());
- ReplicaRequest request = MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
+ ReadWriteMultiRowReplicaRequest request = MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
Review Comment:
Why?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -261,6 +261,35 @@ private void onReplicaMessageReceived(NetworkMessage message, String senderConsi
}
}
+ if (res.repFuture() != null) {
+ assert request instanceof PrimaryReplicaRequest;
+
+ res.repFuture().handle((res0, ex0) -> {
+ NetworkMessage msg0;
+
+ LOG.debug("Sending delayed response for replica request [request={}]", request);
+
+ if (ex == null) {
+ msg0 = prepareReplicaResponse(finalSendTimestamp, res0);
+ } else {
+ LOG.warn("Failed to process delayed response [request={}]", ex, request);
+
+ msg0 = prepareReplicaErrorResponse(finalSendTimestamp, ex);
Review Comment:
As was mentioned above, it's either void or exception, no need to send the result one more time.
##########
modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java:
##########
@@ -365,6 +370,7 @@ public void before() throws Exception {
if (startClient()) {
clientTxManager = new TxManagerImpl(clientReplicaSvc, new HeapLockManager(), clientClock, new TransactionIdGenerator(-1));
+ clientTxManager.start();
Review Comment:
Do we also call stop()? BTW I believe that we do in main, so it worth merging it first.
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java:
##########
@@ -1167,9 +1168,9 @@ private void testWriteIntentOnPrimaryReplica(
raftClientFutureClosure = cmd -> writeFut;
try {
- CompletableFuture<?> replicaWriteFut = partitionReplicaListener.invoke(updatingRequestSupplier.get());
+ CompletableFuture<ReplicaResult> replicaWriteFut = partitionReplicaListener.invoke(updatingRequestSupplier.get());
- assertFalse(replicaWriteFut.isDone());
+ assertTrue(replicaWriteFut.isDone());
Review Comment:
Why?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaResult.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.replicator;
+
+import java.util.concurrent.CompletableFuture;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Represents replica execution result.
+ */
+public class ReplicaResult {
Review Comment:
According to the comments above, I don't think that ReplicaResult should have given structure.
##########
modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestThreeNodesThreeReplicas.java:
##########
@@ -59,4 +69,36 @@ public void after() throws Exception {
super.after();
}
}
+
+ @Test
+ public void testPrimaryReplicaDirectUpdateForExplicitTxn() throws InterruptedException {
+ Peer leader = accRaftClients.get(0).leader();
+ JraftServerImpl server = (JraftServerImpl) raftServers.get(leader.consistentId()).server();
+ var groupId = new TablePartitionId(accounts.tableId(), 0);
+
+ // BLock replication messages to both replicas.
+ server.blockMessages(new RaftNodeId(groupId, leader), (msg, peerId) -> {
+ if (msg instanceof RpcRequests.AppendEntriesRequest) {
+ RpcRequests.AppendEntriesRequest tmp = (AppendEntriesRequest) msg;
+
+ if (tmp.entriesList() != null && !tmp.entriesList().isEmpty()) {
+ return true;
+ }
+ }
+ return false;
+ });
+
+ assertTrue(IgniteTestUtils.waitForCondition(() -> server.blockedMessages(new RaftNodeId(groupId, leader)).size() == 2, 10000),
+ "Failed to wait for blocked messages");
+
+ ReadWriteTransactionImpl tx = (ReadWriteTransactionImpl) igniteTransactions.begin();
+ CompletableFuture<Void> fut = accounts.recordView().upsertAsync(tx, makeValue(1, 100.));
+ // Update must complete now despite the blocked replication protocol.
+ assertTrue(IgniteTestUtils.waitForCondition(fut::isDone, 5_000), "The update future is not completed within timeout");
+
Review Comment:
Do you have an idea of how to check that transaction won't finish until unblocking?
##########
modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestThreeNodesThreeReplicas.java:
##########
@@ -59,4 +69,36 @@ public void after() throws Exception {
super.after();
}
}
+
+ @Test
+ public void testPrimaryReplicaDirectUpdateForExplicitTxn() throws InterruptedException {
+ Peer leader = accRaftClients.get(0).leader();
Review Comment:
It's a bit unreliable. client cache leader, so that actual leader may be on another node. The chance isn't that big, however it's extremely difficult to debug such failures. So I believe that we can ensure that leader is actually the one that we've blocked after the block and either fail the test with meaningful message in order to ease the analysis or even re-block with true leader in a while loop until success. It's also required to check whether the leader was changed or not during the test. Crappy stuff.
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockManager.java:
##########
@@ -33,7 +33,6 @@ public interface LockManager {
* @param lockKey The key.
* @param lockMode Lock mode, for example shared, exclusive, intention-shared etc.
* @return The future with gained lock that will be completed when a lock is successfully acquired.
- * @throws LockException When a lock can't be taken due to possible deadlock.
Review Comment:
Why?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -261,6 +261,37 @@ private void onReplicaMessageReceived(NetworkMessage message, String senderConsi
}
}
+ if (res.repFuture() != null) {
+ assert request instanceof PrimaryReplicaRequest;
+
+ LOG.info("Procesing delayed response for replica request [request={}]", request);
+
+ res.repFuture().handle((res0, ex0) -> {
+ NetworkMessage msg0;
+
+ LOG.info("Sending delayed response for replica request [request={}]", request);
+
+ if (ex == null) {
+ msg0 = prepareReplicaResponse(finalSendTimestamp, res0);
+ } else {
+ LOG.warn("Failed to process delayed response [request={}]", ex, request);
+
+ msg0 = prepareReplicaErrorResponse(finalSendTimestamp, ex);
+ }
+
+ ClusterNode sender = clusterNetSvc.topologyService().getByConsistentId(senderConsistentId);
+
+ if (sender != null) {
+ // Using strong send here is important to avoid a reordering with a normal response.
+ clusterNetSvc.messagingService().send(sender, msg0);
Review Comment:
Agree. I'd rather add operation id or similar same for both fromPrimary and delayed responses in order to match each other. Such ids should be unique within transaction, so basically it's operation counter or similar. WDYT?
--
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