You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "rpuch (via GitHub)" <gi...@apache.org> on 2023/08/15 15:37:50 UTC
[GitHub] [ignite-3] rpuch opened a new pull request, #2450: IGNITE-19227 Wait for schema awailability out of JRaft threads
rpuch opened a new pull request, #2450:
URL: https://github.com/apache/ignite-3/pull/2450
https://issues.apache.org/jira/browse/IGNITE-19227
Thank you for submitting the pull request.
To streamline the review process of the patch and ensure better code quality
we ask both an author and a reviewer to verify the following:
### The Review Checklist
- [ ] **Formal criteria:** TC status, codestyle, mandatory documentation. Also make sure to complete the following:
\- There is a single JIRA ticket related to the pull request.
\- The web-link to the pull request is attached to the JIRA ticket.
\- The JIRA ticket has the Patch Available state.
\- The description of the JIRA ticket explains WHAT was made, WHY and HOW.
\- The pull request title is treated as the final commit message. The following pattern must be used: IGNITE-XXXX Change summary where XXXX - number of JIRA issue.
- [ ] **Design:** new code conforms with the design principles of the components it is added to.
- [ ] **Patch quality:** patch cannot be split into smaller pieces, its size must be reasonable.
- [ ] **Code quality:** code is clean and readable, necessary developer documentation is added if needed.
- [ ] **Tests code quality:** test set covers positive/negative scenarios, happy/edge cases. Tests are effective in terms of execution time and resources.
### Notes
- [Apache Ignite Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide)
--
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] ibessonov commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1299902458
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1158,25 +1169,31 @@ private CompletableFuture<Object> finishTransaction(List<TablePartitionId> aggre
HybridTimestamp currentTimestamp = hybridClock.now();
HybridTimestamp commitTimestamp = commit ? currentTimestamp : null;
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(currentTimestamp.longValue())
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
+ return catalogVersionFor(currentTimestamp)
+ .thenApply(catalogVersion -> {
Review Comment:
How can we guarantee that these asynchronous callbacks won't be reordered? Safe time must be monotonous. I feel like this code is incorrect. Same applies to the rest of the new async code in this class.
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1300200523
##########
modules/core/src/main/java/org/apache/ignite/internal/util/VarIntUtils.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.util;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Utilities to work with general-purpose varints.
+ */
+public class VarIntUtils {
Review Comment:
I don't think this should be done in this PR. Touching V1Decoder/V1Encoder would at least require to write some tests for them (that are absent). I suggest to file a separate issue for this.
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1302642650
##########
modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/TestIgnitionManager.java:
##########
@@ -100,20 +104,35 @@ private static InitParameters applyTestDefaultsToClusterConfig(InitParameters pa
.metaStorageNodeNames(params.metaStorageNodeNames())
.cmgNodeNames(params.cmgNodeNames());
+ ConfigDocument configDocument;
+
if (params.clusterConfiguration() == null) {
- builder.clusterConfiguration("{ schemaSync.delayDuration: 0 }");
+ configDocument = ConfigDocumentFactory.parseString("{}");
} else {
- ConfigDocument configDocument = ConfigDocumentFactory.parseString(params.clusterConfiguration());
+ configDocument = ConfigDocumentFactory.parseString(params.clusterConfiguration());
+ }
- String delayDurationPath = "schemaSync.delayDuration";
+ configDocument = applyTestDefault(
Review Comment:
If someone wants to run a cluster with a different value of some cluster-wide config property, they would use `TestIgnitionManager#init()` passing the config they need. The code we are looking at now is about DEFAULT values used if no explicit value was supplied when initializing the cluster.
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301697545
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/CatalogVersionAware.java:
##########
@@ -0,0 +1,28 @@
+/*
+ * 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.command;
+
+/**
+ * A command that requires certain level of catalog version to be locally available just to be accepted on the node.
+ */
+public interface CatalogVersionAware {
Review Comment:
It should not be used here because it's not something that is designed to be used as a lambda or a method reference. This is just a mixin interface having exactly one method.
--
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] alievmirza commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "alievmirza (via GitHub)" <gi...@apache.org>.
alievmirza commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1307038374
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnAppendEntries.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.raft.jraft.Node;
+import org.apache.ignite.raft.jraft.entity.EnumOutter.EntryType;
+import org.apache.ignite.raft.jraft.entity.RaftOutter;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.EntryMeta;
+import org.apache.ignite.raft.jraft.error.RaftError;
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.rpc.RaftRpcFactory;
+import org.apache.ignite.raft.jraft.rpc.RaftServerService;
+import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
+import org.apache.ignite.raft.jraft.rpc.impl.core.AppendEntriesRequestInterceptor;
+import org.apache.ignite.raft.jraft.util.Marshaller;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An {@link AppendEntriesRequestInterceptor} that rejects requests (by returning EBUSY error code) if any of the
+ * incoming commands requires catalog version that is not available locally yet.
+ */
+public class CheckCatalogVersionOnAppendEntries implements AppendEntriesRequestInterceptor {
+ private static final IgniteLogger LOG = Loggers.forClass(CheckCatalogVersionOnAppendEntries.class);
+
+ private static final int NO_VERSION_REQUIREMENT = Integer.MIN_VALUE;
+
+ private final CatalogService catalogService;
+
+ public CheckCatalogVersionOnAppendEntries(CatalogService catalogService) {
+ this.catalogService = catalogService;
+ }
+
+ @Override
+ public @Nullable Message intercept(RaftServerService service, AppendEntriesRequest request, RpcRequestClosure done) {
+ if (request.entriesList() == null || request.data() == null) {
+ return null;
+ }
+
+ Node node = (Node) service;
+
+ ByteBuffer allData = request.data().asReadOnlyByteBuffer();
+ int offset = 0;
+
+ for (RaftOutter.EntryMeta entry : request.entriesList()) {
+ int requiredCatalogVersion = readRequiredCatalogVersionForMeta(allData, entry, node.getOptions().requiredCommandsMarshaller());
+
+ if (requiredCatalogVersion != NO_VERSION_REQUIREMENT && !isMetadataAvailableFor(requiredCatalogVersion)) {
+ LOG.warn("Metadata not yet available, group {}, required level {}.", request.groupId(), requiredCatalogVersion);
+ return RaftRpcFactory.DEFAULT //
+ .newResponse(node.getRaftOptions().getRaftMessagesFactory(), RaftError.EBUSY,
+ "Metadata not yet available, group '%s', required level %d.", request.groupId(), requiredCatalogVersion);
+ }
+
+ offset += (int) entry.dataLen();
+ allData.position(offset);
+ }
+
+ return null;
+ }
+
+ private static int readRequiredCatalogVersionForMeta(ByteBuffer allData, final EntryMeta entry, Marshaller commandsMarshaller) {
+ if (entry.type() != EntryType.ENTRY_TYPE_DATA) {
+ return NO_VERSION_REQUIREMENT;
+ }
+
+ if (!(commandsMarshaller instanceof PartitionCommandsMarshaller)) {
+ return NO_VERSION_REQUIREMENT;
+ }
+
+ PartitionCommandsMarshaller partitionCommandsMarshaller = (PartitionCommandsMarshaller) commandsMarshaller;
+
+ long dataLen = entry.dataLen();
+ if (dataLen > 0) {
+ return partitionCommandsMarshaller.readRequiredCatalogVersion(allData);
+ }
+
+ return NO_VERSION_REQUIREMENT;
+ }
+
+ private boolean isMetadataAvailableFor(int requiredCatalogVersion) {
+ return requiredCatalogVersion <= catalogService.latestCatalogVersion();
Review Comment:
I'm a bit confused with the javadoc of `CatalogService#latestCatalogVersion`
> <p>NOTE: This method should only be used at the start of components that may be removed or moved in the future.
which is not true anymore
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnAppendEntries.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.raft.jraft.Node;
+import org.apache.ignite.raft.jraft.entity.EnumOutter.EntryType;
+import org.apache.ignite.raft.jraft.entity.RaftOutter;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.EntryMeta;
+import org.apache.ignite.raft.jraft.error.RaftError;
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.rpc.RaftRpcFactory;
+import org.apache.ignite.raft.jraft.rpc.RaftServerService;
+import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
+import org.apache.ignite.raft.jraft.rpc.impl.core.AppendEntriesRequestInterceptor;
+import org.apache.ignite.raft.jraft.util.Marshaller;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An {@link AppendEntriesRequestInterceptor} that rejects requests (by returning EBUSY error code) if any of the
+ * incoming commands requires catalog version that is not available locally yet.
+ */
+public class CheckCatalogVersionOnAppendEntries implements AppendEntriesRequestInterceptor {
+ private static final IgniteLogger LOG = Loggers.forClass(CheckCatalogVersionOnAppendEntries.class);
+
+ private static final int NO_VERSION_REQUIREMENT = Integer.MIN_VALUE;
+
+ private final CatalogService catalogService;
+
+ public CheckCatalogVersionOnAppendEntries(CatalogService catalogService) {
+ this.catalogService = catalogService;
+ }
+
+ @Override
+ public @Nullable Message intercept(RaftServerService service, AppendEntriesRequest request, RpcRequestClosure done) {
+ if (request.entriesList() == null || request.data() == null) {
+ return null;
+ }
+
+ Node node = (Node) service;
+
+ ByteBuffer allData = request.data().asReadOnlyByteBuffer();
+ int offset = 0;
+
+ for (RaftOutter.EntryMeta entry : request.entriesList()) {
+ int requiredCatalogVersion = readRequiredCatalogVersionForMeta(allData, entry, node.getOptions().requiredCommandsMarshaller());
+
+ if (requiredCatalogVersion != NO_VERSION_REQUIREMENT && !isMetadataAvailableFor(requiredCatalogVersion)) {
+ LOG.warn("Metadata not yet available, group {}, required level {}.", request.groupId(), requiredCatalogVersion);
+ return RaftRpcFactory.DEFAULT //
+ .newResponse(node.getRaftOptions().getRaftMessagesFactory(), RaftError.EBUSY,
+ "Metadata not yet available, group '%s', required level %d.", request.groupId(), requiredCatalogVersion);
+ }
+
+ offset += (int) entry.dataLen();
+ allData.position(offset);
+ }
+
+ return null;
+ }
+
+ private static int readRequiredCatalogVersionForMeta(ByteBuffer allData, final EntryMeta entry, Marshaller commandsMarshaller) {
+ if (entry.type() != EntryType.ENTRY_TYPE_DATA) {
+ return NO_VERSION_REQUIREMENT;
+ }
+
+ if (!(commandsMarshaller instanceof PartitionCommandsMarshaller)) {
+ return NO_VERSION_REQUIREMENT;
+ }
+
+ PartitionCommandsMarshaller partitionCommandsMarshaller = (PartitionCommandsMarshaller) commandsMarshaller;
+
+ long dataLen = entry.dataLen();
+ if (dataLen > 0) {
+ return partitionCommandsMarshaller.readRequiredCatalogVersion(allData);
+ }
+
+ return NO_VERSION_REQUIREMENT;
+ }
+
+ private boolean isMetadataAvailableFor(int requiredCatalogVersion) {
+ return requiredCatalogVersion <= catalogService.latestCatalogVersion();
Review Comment:
I'm a bit confused with the javadoc of `CatalogService#latestCatalogVersion`:
> <p>NOTE: This method should only be used at the start of components that may be removed or moved in the future.
which is not true anymore
--
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 #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1305631114
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1158,25 +1169,31 @@ private CompletableFuture<Object> finishTransaction(List<TablePartitionId> aggre
HybridTimestamp currentTimestamp = hybridClock.now();
HybridTimestamp commitTimestamp = commit ? currentTimestamp : null;
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(currentTimestamp.longValue())
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
+ return catalogVersionFor(currentTimestamp)
+ .thenApply(catalogVersion -> {
Review Comment:
We will fix it within another task (GNITE-20116), and as long as this issue is not processed, we will not think about synchronization.
I have no complaints against this patch.
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301691368
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/schemasync/ItSchemaSyncAndReplicationTest.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.schemasync;
+
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.ReplicationGroupsUtils;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.metastorage.server.raft.MetastorageGroupId;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.schema.CheckCatalogVersionOnAppendEntries;
+import org.apache.ignite.internal.test.WatchListenerInhibitor;
+import org.apache.ignite.internal.testframework.jul.NoOpHandler;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests about interaction between Schema Synchronization and Replication.
+ */
+@SuppressWarnings("resource")
+class ItSchemaSyncAndReplicationTest extends ClusterPerTestIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 3;
+ }
+
+ /**
+ * The replication mechanism must not replicate commands for which schemas are not yet available on the node
+ * to which replication happens (in Raft, it means that followers/learners cannot receive commands that they
+ * cannot execute without waiting for schemas). This method tests this scenario.
+ */
+ @Test
+ void laggingSchemasPreventPartitionDataReplication() throws Exception {
+ createTestTableWith3Replicas();
+
+ final int notInhibitedNodeIndex = 0;
+ transferLeadershipsTo(notInhibitedNodeIndex);
+
+ IgniteImpl nodeToInhibitMetaStorage = cluster.node(1);
+
+ WatchListenerInhibitor listenerInhibitor = WatchListenerInhibitor.metastorageEventsInhibitor(nodeToInhibitMetaStorage);
+ listenerInhibitor.startInhibit();
+
+ try {
+ CountDownLatch rejectionTriggered = rejectionDueToMetadataLagTriggered();
+
+ updateTableSchemaAt(notInhibitedNodeIndex);
+ putToTableAt(notInhibitedNodeIndex);
+
+ assertTrue(rejectionTriggered.await(10, TimeUnit.SECONDS), "Did not see rejections due to lagging metadata");
+
+ assertTrue(solePartitionIsEmpty(nodeToInhibitMetaStorage), "Something was written to the partition");
+
+ listenerInhibitor.stopInhibit();
+
+ assertTrue(
+ waitForCondition(() -> !solePartitionIsEmpty(nodeToInhibitMetaStorage), 10_000),
+ "Nothing was written to partition even after inhibiting was cancelled"
+ );
+ } finally {
+ listenerInhibitor.stopInhibit();
+ }
+ }
+
+ private void createTestTableWith3Replicas() throws InterruptedException {
+ String zoneSql = "create zone test_zone with partitions=1, replicas=3";
+ String sql = "create table test (key int primary key, value varchar(20))"
+ + " with primary_zone='TEST_ZONE'";
+
+ cluster.doInSession(0, session -> {
+ executeUpdate(zoneSql, session);
+ executeUpdate(sql, session);
+ });
+
+ waitForTableToStart();
+ }
+
+ private void waitForTableToStart() throws InterruptedException {
+ // TODO: IGNITE-18733 - remove this wait because when a table creation query is executed, the table must be fully ready.
+
+ BooleanSupplier tableStarted = () -> {
+ int numberOfStartedRaftNodes = cluster.runningNodes()
+ .map(ReplicationGroupsUtils::tablePartitionIds)
+ .mapToInt(List::size)
+ .sum();
+ return numberOfStartedRaftNodes == 3;
+ };
+
+ assertTrue(waitForCondition(tableStarted, 10_000), "Did not see all table RAFT nodes started");
+ }
+
+ private void transferLeadershipsTo(int nodeIndex) throws InterruptedException {
+ cluster.transferLeadershipTo(nodeIndex, MetastorageGroupId.INSTANCE);
+ cluster.transferLeadershipTo(nodeIndex, cluster.solePartitionId());
+ }
+
+ private static CountDownLatch rejectionDueToMetadataLagTriggered() {
+ Logger interceptorLogger = Logger.getLogger(CheckCatalogVersionOnAppendEntries.class.getName());
+
+ CountDownLatch rejectionTriggered = new CountDownLatch(1);
Review Comment:
Both future and latch allow to wait for some event to happen, but the latch seems to be more focused on this task. On the other hand, what's the advantage of a future over a latch?
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1300153908
##########
modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/TestIgnitionManager.java:
##########
@@ -100,20 +104,35 @@ private static InitParameters applyTestDefaultsToClusterConfig(InitParameters pa
.metaStorageNodeNames(params.metaStorageNodeNames())
.cmgNodeNames(params.cmgNodeNames());
+ ConfigDocument configDocument;
+
if (params.clusterConfiguration() == null) {
- builder.clusterConfiguration("{ schemaSync.delayDuration: 0 }");
+ configDocument = ConfigDocumentFactory.parseString("{}");
} else {
- ConfigDocument configDocument = ConfigDocumentFactory.parseString(params.clusterConfiguration());
+ configDocument = ConfigDocumentFactory.parseString(params.clusterConfiguration());
+ }
- String delayDurationPath = "schemaSync.delayDuration";
+ configDocument = applyTestDefault(
Review Comment:
It's marked `@Immutable` in the configuration, so there should not be any way to override it yet
--
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] ascherbakoff commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "ascherbakoff (via GitHub)" <gi...@apache.org>.
ascherbakoff commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1307247481
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/IgniteRpcServer.java:
##########
@@ -52,10 +51,10 @@
import org.apache.ignite.raft.jraft.rpc.impl.cli.ResetPeerRequestProcessor;
import org.apache.ignite.raft.jraft.rpc.impl.cli.SnapshotRequestProcessor;
import org.apache.ignite.raft.jraft.rpc.impl.cli.TransferLeaderRequestProcessor;
-import org.apache.ignite.raft.jraft.rpc.impl.core.AppendEntriesRequestProcessor;
+import org.apache.ignite.raft.jraft.rpc.impl.core.AppendEntriesRequestInterceptor;import org.apache.ignite.raft.jraft.rpc.impl.core.AppendEntriesRequestProcessor;
Review Comment:
Why two imports on one line ?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java:
##########
@@ -91,6 +87,8 @@ public void handleRequest(RpcContext rpcCtx, ActionRequest request) {
* @param rpcCtx The context.
*/
private void applyWrite(Node node, ActionRequest request, RpcContext rpcCtx) {
+ Marshaller commandsMarshaller = node.getOptions().requiredCommandsMarshaller();
Review Comment:
Checking each time for null doesn't make sense. A check can be done once on init.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnAppendEntries.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.raft.jraft.Node;
+import org.apache.ignite.raft.jraft.entity.EnumOutter.EntryType;
+import org.apache.ignite.raft.jraft.entity.RaftOutter;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.EntryMeta;
+import org.apache.ignite.raft.jraft.error.RaftError;
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.rpc.RaftRpcFactory;
+import org.apache.ignite.raft.jraft.rpc.RaftServerService;
+import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
+import org.apache.ignite.raft.jraft.rpc.impl.core.AppendEntriesRequestInterceptor;
+import org.apache.ignite.raft.jraft.util.Marshaller;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An {@link AppendEntriesRequestInterceptor} that rejects requests (by returning EBUSY error code) if any of the
+ * incoming commands requires catalog version that is not available locally yet.
+ */
+public class CheckCatalogVersionOnAppendEntries implements AppendEntriesRequestInterceptor {
+ private static final IgniteLogger LOG = Loggers.forClass(CheckCatalogVersionOnAppendEntries.class);
+
+ private static final int NO_VERSION_REQUIREMENT = Integer.MIN_VALUE;
+
+ private final CatalogService catalogService;
+
+ public CheckCatalogVersionOnAppendEntries(CatalogService catalogService) {
+ this.catalogService = catalogService;
+ }
+
+ @Override
+ public @Nullable Message intercept(RaftServerService service, AppendEntriesRequest request, RpcRequestClosure done) {
+ if (request.entriesList() == null || request.data() == null) {
+ return null;
+ }
+
+ Node node = (Node) service;
+
+ ByteBuffer allData = request.data().asReadOnlyByteBuffer();
+ int offset = 0;
+
+ for (RaftOutter.EntryMeta entry : request.entriesList()) {
+ int requiredCatalogVersion = readRequiredCatalogVersionForMeta(allData, entry, node.getOptions().requiredCommandsMarshaller());
+
+ if (requiredCatalogVersion != NO_VERSION_REQUIREMENT && !isMetadataAvailableFor(requiredCatalogVersion)) {
+ LOG.warn("Metadata not yet available, group {}, required level {}.", request.groupId(), requiredCatalogVersion);
+ return RaftRpcFactory.DEFAULT //
Review Comment:
We need to keep an eye on a number of rejected messages, some new stat field would be great. Can be done as a separate ticket.
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftGroupOptions.java:
##########
@@ -42,6 +43,9 @@ public class RaftGroupOptions {
/** Configuration of own striped disruptor for FSMCaller service of raft node, {@code null} means use shared disruptor. */
private @Nullable RaftNodeDisruptorConfiguration ownFsmCallerExecutorDisruptorConfig;
+ /** Marshaller to marshall/unmarshall commands. */
+ private @Nullable Marshaller commandsMarshaller;
Review Comment:
Why do we need to override marshaller ? For test purposes ?
##########
modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java:
##########
@@ -513,12 +517,18 @@ public class IgniteImpl implements Ignite {
SchemaSynchronizationConfiguration.KEY
);
+ LongSupplier delayDurationMsSupplier = () -> schemaSyncConfig.delayDuration().value();
+
catalogManager = new CatalogManagerImpl(
new UpdateLogImpl(metaStorageMgr),
clockWaiter,
- () -> schemaSyncConfig.delayDuration().value()
+ delayDurationMsSupplier
Review Comment:
Why not pass a constant instead of closure ?
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1307705675
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnAppendEntries.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.raft.jraft.Node;
+import org.apache.ignite.raft.jraft.entity.EnumOutter.EntryType;
+import org.apache.ignite.raft.jraft.entity.RaftOutter;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.EntryMeta;
+import org.apache.ignite.raft.jraft.error.RaftError;
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.rpc.RaftRpcFactory;
+import org.apache.ignite.raft.jraft.rpc.RaftServerService;
+import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
+import org.apache.ignite.raft.jraft.rpc.impl.core.AppendEntriesRequestInterceptor;
+import org.apache.ignite.raft.jraft.util.Marshaller;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An {@link AppendEntriesRequestInterceptor} that rejects requests (by returning EBUSY error code) if any of the
+ * incoming commands requires catalog version that is not available locally yet.
+ */
+public class CheckCatalogVersionOnAppendEntries implements AppendEntriesRequestInterceptor {
+ private static final IgniteLogger LOG = Loggers.forClass(CheckCatalogVersionOnAppendEntries.class);
+
+ private static final int NO_VERSION_REQUIREMENT = Integer.MIN_VALUE;
+
+ private final CatalogService catalogService;
+
+ public CheckCatalogVersionOnAppendEntries(CatalogService catalogService) {
+ this.catalogService = catalogService;
+ }
+
+ @Override
+ public @Nullable Message intercept(RaftServerService service, AppendEntriesRequest request, RpcRequestClosure done) {
+ if (request.entriesList() == null || request.data() == null) {
+ return null;
+ }
+
+ Node node = (Node) service;
+
+ ByteBuffer allData = request.data().asReadOnlyByteBuffer();
+ int offset = 0;
+
+ for (RaftOutter.EntryMeta entry : request.entriesList()) {
+ int requiredCatalogVersion = readRequiredCatalogVersionForMeta(allData, entry, node.getOptions().requiredCommandsMarshaller());
+
+ if (requiredCatalogVersion != NO_VERSION_REQUIREMENT && !isMetadataAvailableFor(requiredCatalogVersion)) {
+ LOG.warn("Metadata not yet available, group {}, required level {}.", request.groupId(), requiredCatalogVersion);
+ return RaftRpcFactory.DEFAULT //
Review Comment:
I filed a ticket https://issues.apache.org/jira/browse/IGNITE-20298 and added a TODO mentioning this ticket
--
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] ibessonov commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1302604986
##########
modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/TestIgnitionManager.java:
##########
@@ -100,20 +104,35 @@ private static InitParameters applyTestDefaultsToClusterConfig(InitParameters pa
.metaStorageNodeNames(params.metaStorageNodeNames())
.cmgNodeNames(params.cmgNodeNames());
+ ConfigDocument configDocument;
+
if (params.clusterConfiguration() == null) {
- builder.clusterConfiguration("{ schemaSync.delayDuration: 0 }");
+ configDocument = ConfigDocumentFactory.parseString("{}");
} else {
- ConfigDocument configDocument = ConfigDocumentFactory.parseString(params.clusterConfiguration());
+ configDocument = ConfigDocumentFactory.parseString(params.clusterConfiguration());
+ }
- String delayDurationPath = "schemaSync.delayDuration";
+ configDocument = applyTestDefault(
Review Comment:
I mean, what if subclass wants to have its own value?
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1300181243
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1158,25 +1169,31 @@ private CompletableFuture<Object> finishTransaction(List<TablePartitionId> aggre
HybridTimestamp currentTimestamp = hybridClock.now();
HybridTimestamp commitTimestamp = commit ? currentTimestamp : null;
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(currentTimestamp.longValue())
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
+ return catalogVersionFor(currentTimestamp)
+ .thenApply(catalogVersion -> {
+ FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
+ .txId(txId)
+ .commit(commit)
+ .safeTimeLong(currentTimestamp.longValue())
+ .requiredCatalogVersion(catalogVersion)
+ .tablePartitionIds(
+ aggregatedGroupIds.stream()
+ .map(PartitionReplicaListener::tablePartitionId)
+ .collect(toList())
+ );
+
+ if (commit) {
+ finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
Review Comment:
It matches the safeTime only for commits; for aborts it's null
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1302645629
##########
modules/core/src/main/java/org/apache/ignite/internal/util/VarIntUtils.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.util;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Utilities to work with general-purpose varints.
+ */
+public class VarIntUtils {
Review Comment:
Filed an issue: https://issues.apache.org/jira/browse/IGNITE-20268
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301563288
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -122,10 +125,12 @@ public class JraftServerImpl implements RaftServer {
private ExecutorService requestExecutor;
/** Marshaller for RAFT commands. */
- private final Marshaller commandsMarshaller;
+ private final Marshaller defaultCommandsMarshaller;
Review Comment:
Because it's used if not overriden in group options
--
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] ibessonov commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1299791137
##########
modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/TestIgnitionManager.java:
##########
@@ -100,20 +104,35 @@ private static InitParameters applyTestDefaultsToClusterConfig(InitParameters pa
.metaStorageNodeNames(params.metaStorageNodeNames())
.cmgNodeNames(params.cmgNodeNames());
+ ConfigDocument configDocument;
+
if (params.clusterConfiguration() == null) {
- builder.clusterConfiguration("{ schemaSync.delayDuration: 0 }");
+ configDocument = ConfigDocumentFactory.parseString("{}");
} else {
- ConfigDocument configDocument = ConfigDocumentFactory.parseString(params.clusterConfiguration());
+ configDocument = ConfigDocumentFactory.parseString(params.clusterConfiguration());
+ }
- String delayDurationPath = "schemaSync.delayDuration";
+ configDocument = applyTestDefault(
Review Comment:
There's no way to override this value, right?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java:
##########
@@ -91,7 +87,7 @@ public void handleRequest(RpcContext rpcCtx, ActionRequest request) {
* @param rpcCtx The context.
*/
private void applyWrite(Node node, ActionRequest request, RpcContext rpcCtx) {
- node.apply(new Task(ByteBuffer.wrap(commandsMarshaller.marshall(request.command())),
+ node.apply(new Task(ByteBuffer.wrap(node.getOptions().requiredCommandsMarshaller().marshall(request.command())),
Review Comment:
Please extract a variable, the line is pretty long already :)
##########
modules/core/src/main/java/org/apache/ignite/internal/util/VarIntUtils.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.util;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Utilities to work with general-purpose varints.
+ */
+public class VarIntUtils {
Review Comment:
I completely forgot that such code is also duplicated in V1Decoder / V1Encoder. Please simplify that code as well.
The only real difference is that we use "long" there, but data format is the same.
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java:
##########
@@ -146,6 +147,16 @@ public Loza(
);
}
+ /**
+ * Sets {@link AppendEntriesRequestInterceptor} to use.
+ *
+ * @param appendEntriesRequestInterceptor Interceptor to use.
+ */
+ public void appendEntriesRequestInterceptor(AppendEntriesRequestInterceptor appendEntriesRequestInterceptor) {
Review Comment:
Can we pass it into a constructor?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/PartitionCommandsMarshallerImpl.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.raft.util.OptimizedMarshaller;
+import org.apache.ignite.internal.table.distributed.command.CatalogLevelAware;
+import org.apache.ignite.internal.util.VarIntUtils;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+
+/**
+ * Default {@link PartitionCommandsMarshaller} implementation.
+ */
+public class PartitionCommandsMarshallerImpl extends OptimizedMarshaller implements PartitionCommandsMarshaller {
+ public PartitionCommandsMarshallerImpl(MessageSerializationRegistry serializationRegistry) {
+ super(serializationRegistry);
+ }
+
+ @Override
+ public byte[] marshall(Object o) {
+ int requiredCatalogVersion = o instanceof CatalogLevelAware ? ((CatalogLevelAware) o).requiredCatalogVersion() : 0;
Review Comment:
Should it be -1 otherwise, or it doesn't matter?
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/util/OptimizedMarshaller.java:
##########
@@ -102,7 +100,11 @@ public byte[] marshall(Object o) {
buffer = expandBuffer(buffer);
}
- return Arrays.copyOf(buffer.array(), buffer.position());
+ byte[] result = Arrays.copyOf(buffer.array(), buffer.position());
+
+ buffer.position(0);
Review Comment:
Please assert that position is 0 in constructor
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/CatalogLevelAware.java:
##########
@@ -0,0 +1,28 @@
+/*
+ * 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.command;
+
+/**
+ * A command that requires certain level of catalog version to be locally available just to be accepted on the node.
+ */
+public interface CatalogLevelAware {
Review Comment:
Maybe it should be "catalog version", as everywhere else?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1158,25 +1169,31 @@ private CompletableFuture<Object> finishTransaction(List<TablePartitionId> aggre
HybridTimestamp currentTimestamp = hybridClock.now();
HybridTimestamp commitTimestamp = commit ? currentTimestamp : null;
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(currentTimestamp.longValue())
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
+ return catalogVersionFor(currentTimestamp)
+ .thenApply(catalogVersion -> {
Review Comment:
How can we guarantee that these asynchronous callbacks won't be reordered? Safe time must be monotonous. I fell like this code is incorrect. Same applies to the rest of the new async code in this class.
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/schema/PartitionCommandsMarshallerImplTest.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.schema;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+import org.apache.ignite.internal.raft.util.OptimizedMarshaller;
+import org.apache.ignite.internal.replicator.command.SafeTimeSyncCommand;
+import org.apache.ignite.internal.replicator.command.SafeTimeSyncCommandSerializationFactory;
+import org.apache.ignite.internal.replicator.message.ReplicaMessageGroup;
+import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
+import org.apache.ignite.internal.table.distributed.TableMessageGroup;
+import org.apache.ignite.internal.table.distributed.TableMessagesFactory;
+import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
+import org.apache.ignite.internal.table.distributed.command.FinishTxCommandSerializationFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.junit.jupiter.api.Test;
+
+class PartitionCommandsMarshallerImplTest {
+ private final TableMessagesFactory tableMessagesFactory = new TableMessagesFactory();
+ private final ReplicaMessagesFactory replicaMessagesFactory = new ReplicaMessagesFactory();
+
+ private final MessageSerializationRegistry registry = new MessageSerializationRegistryImpl();
+
+ {
+ // For a command that has required catalog version property.
+ registry.registerFactory(
+ TableMessageGroup.GROUP_TYPE,
+ TableMessageGroup.Commands.FINISH_TX,
+ new FinishTxCommandSerializationFactory(tableMessagesFactory)
+ );
+
+ // For a command that does not have required catalog version property.
+ registry.registerFactory(
+ (short) 8,
Review Comment:
Is there a constant for this that you could use?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnAppendEntries.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.raft.jraft.Node;
+import org.apache.ignite.raft.jraft.entity.EnumOutter.EntryType;
+import org.apache.ignite.raft.jraft.entity.RaftOutter;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.EntryMeta;
+import org.apache.ignite.raft.jraft.error.RaftError;
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.rpc.RaftRpcFactory;
+import org.apache.ignite.raft.jraft.rpc.RaftServerService;
+import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
+import org.apache.ignite.raft.jraft.rpc.impl.AppendEntriesRequestInterceptor;
+import org.apache.ignite.raft.jraft.util.Marshaller;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An {@link AppendEntriesRequestInterceptor} that rejects requests (by returning EBUSY error code) if any of the
+ * incoming commands requires catalog version that is not available locally yet.
+ */
+public class CheckCatalogVersionOnAppendEntries implements AppendEntriesRequestInterceptor {
+ private static final IgniteLogger LOG = Loggers.forClass(CheckCatalogVersionOnAppendEntries.class);
+
+ private static final int NO_LEVEL_REQUIREMENT = Integer.MIN_VALUE;
Review Comment:
Maybe we should call it "version"
##########
modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/TestIgnitionManager.java:
##########
@@ -100,20 +104,35 @@ private static InitParameters applyTestDefaultsToClusterConfig(InitParameters pa
.metaStorageNodeNames(params.metaStorageNodeNames())
.cmgNodeNames(params.cmgNodeNames());
+ ConfigDocument configDocument;
+
if (params.clusterConfiguration() == null) {
- builder.clusterConfiguration("{ schemaSync.delayDuration: 0 }");
+ configDocument = ConfigDocumentFactory.parseString("{}");
} else {
- ConfigDocument configDocument = ConfigDocumentFactory.parseString(params.clusterConfiguration());
+ configDocument = ConfigDocumentFactory.parseString(params.clusterConfiguration());
+ }
- String delayDurationPath = "schemaSync.delayDuration";
+ configDocument = applyTestDefault(
Review Comment:
I wish we would be able to get configuration from annotations, but that's for the future
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java:
##########
@@ -79,6 +79,13 @@ public boolean isSingleRow() {
}
}
+ /**
+ * Returns {@code true} if the operation works with a single row and it's a write.
+ */
+ public boolean isSingleRowWrite() {
Review Comment:
It's only used in a single test, right? Are you sure that we need to pollute this class even more?
I feel like some other methods are also test-only here.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1158,25 +1169,31 @@ private CompletableFuture<Object> finishTransaction(List<TablePartitionId> aggre
HybridTimestamp currentTimestamp = hybridClock.now();
HybridTimestamp commitTimestamp = commit ? currentTimestamp : null;
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(currentTimestamp.longValue())
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
+ return catalogVersionFor(currentTimestamp)
+ .thenApply(catalogVersion -> {
+ FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
+ .txId(txId)
+ .commit(commit)
+ .safeTimeLong(currentTimestamp.longValue())
+ .requiredCatalogVersion(catalogVersion)
+ .tablePartitionIds(
+ aggregatedGroupIds.stream()
+ .map(PartitionReplicaListener::tablePartitionId)
+ .collect(toList())
+ );
+
+ if (commit) {
+ finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
Review Comment:
Off-topic: any idea, why do even need a commit timestamp, if it always matches the safe time?
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301677730
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -122,10 +125,12 @@ public class JraftServerImpl implements RaftServer {
private ExecutorService requestExecutor;
/** Marshaller for RAFT commands. */
- private final Marshaller commandsMarshaller;
+ private final Marshaller defaultCommandsMarshaller;
/** Raft service event interceptor. */
- private RaftServiceEventInterceptor serviceEventInterceptor;
+ private final RaftServiceEventInterceptor serviceEventInterceptor;
+
+ private AppendEntriesRequestInterceptor appendEntriesRequestInterceptor = new NullAppendEntriesRequestInterceptor();
Review Comment:
The setter is called from the thread that starts an Ignite node; the field is accessed in the `start()` method that is called from the same thread, so no concurrency issues are expected here. Do you have any specific scenario in mind?
--
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] ibessonov commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1302659227
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1158,25 +1169,31 @@ private CompletableFuture<Object> finishTransaction(List<TablePartitionId> aggre
HybridTimestamp currentTimestamp = hybridClock.now();
HybridTimestamp commitTimestamp = commit ? currentTimestamp : null;
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(currentTimestamp.longValue())
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
+ return catalogVersionFor(currentTimestamp)
+ .thenApply(catalogVersion -> {
+ FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
+ .txId(txId)
+ .commit(commit)
+ .safeTimeLong(currentTimestamp.longValue())
+ .requiredCatalogVersion(catalogVersion)
+ .tablePartitionIds(
+ aggregatedGroupIds.stream()
+ .map(PartitionReplicaListener::tablePartitionId)
+ .collect(toList())
+ );
+
+ if (commit) {
+ finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
Review Comment:
Ok, so it matches safeTime every time we read it. Looks like an excessive data to me :) We should get rid of it in the future I think
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301112670
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java:
##########
@@ -79,6 +79,13 @@ public boolean isSingleRow() {
}
}
+ /**
+ * Returns {@code true} if the operation works with a single row and it's a write.
+ */
+ public boolean isSingleRowWrite() {
Review Comment:
I mean, `RequestType` is an enum, it had no code before these methods were added. And the methods are about the members of the enum, so it is not unnatural for them to belong to the enum. The only thing is that these methods are only used in tests, but it does not look like a crime.
I would certainly not want to move them to `PartitionReplicaListenerTest` which is already pretty big.
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301698550
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/AppendEntriesRequestInterceptor.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.raft.jraft.rpc.impl;
+
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.rpc.RaftServerService;
+import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Intercepts {@link AppendEntriesRequest}s as they come in. It might be used to handle such a request in a non-standard
+ * way (like returning EBUSY under special circumstances instead of the standard behavior).
+ */
+public interface AppendEntriesRequestInterceptor {
Review Comment:
I doubt it's needed here, there is no intention to use lambda functions to implement this interface.
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1307559792
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftGroupOptions.java:
##########
@@ -42,6 +43,9 @@ public class RaftGroupOptions {
/** Configuration of own striped disruptor for FSMCaller service of raft node, {@code null} means use shared disruptor. */
private @Nullable RaftNodeDisruptorConfiguration ownFsmCallerExecutorDisruptorConfig;
+ /** Marshaller to marshall/unmarshall commands. */
+ private @Nullable Marshaller commandsMarshaller;
Review Comment:
For partitions, we need to quickly extract requiredCatalogVersion from command's serialized representation without parsing the whole command. To do so, I tweaked the marshaller used for partitions, but it adds one byte to each command that does not have that requiredCatalogVersion. If we use same marshaller for other (non-partition) groups, we unnecessarily increase their (serialized) sizes as well. Hence the necessity to have ability to supply a different marshaller.
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1300182786
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java:
##########
@@ -79,6 +79,13 @@ public boolean isSingleRow() {
}
}
+ /**
+ * Returns {@code true} if the operation works with a single row and it's a write.
+ */
+ public boolean isSingleRowWrite() {
Review Comment:
How is this pollution?
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1300160949
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java:
##########
@@ -146,6 +147,16 @@ public Loza(
);
}
+ /**
+ * Sets {@link AppendEntriesRequestInterceptor} to use.
+ *
+ * @param appendEntriesRequestInterceptor Interceptor to use.
+ */
+ public void appendEntriesRequestInterceptor(AppendEntriesRequestInterceptor appendEntriesRequestInterceptor) {
Review Comment:
Unfortunately, there is a cyclic dependency: `Loza` needs the interceptor, which needs `CatalogManager`, which needs `MetaStorageManager`, which needs `Loca`.
--
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] tkalkirill commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301241925
##########
modules/core/src/test/java/org/apache/ignite/internal/util/VarIntUtilsTest.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.util;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import java.nio.ByteBuffer;
+import java.util.stream.Stream;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class VarIntUtilsTest {
Review Comment:
I think that we need a test that will honestly check `VarIntUtils#varIntLength`.
##########
modules/core/src/main/java/org/apache/ignite/internal/util/VarIntUtils.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.util;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Utilities to work with general-purpose varints.
+ */
+public class VarIntUtils {
+ /**
+ * Returns number of bytes that are needed to represent the given integer as a varint.
+ *
+ * @param val Int value.
+ * @return Number of bytes that are needed to represent the given integer as a varint.
Review Comment:
```suggestion
* @param val Int value.
```
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -122,10 +125,12 @@ public class JraftServerImpl implements RaftServer {
private ExecutorService requestExecutor;
/** Marshaller for RAFT commands. */
- private final Marshaller commandsMarshaller;
+ private final Marshaller defaultCommandsMarshaller;
Review Comment:
why default?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/AppendEntriesRequestInterceptor.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.raft.jraft.rpc.impl;
+
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.rpc.RaftServerService;
+import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Intercepts {@link AppendEntriesRequest}s as they come in. It might be used to handle such a request in a non-standard
+ * way (like returning EBUSY under special circumstances instead of the standard behavior).
+ */
+public interface AppendEntriesRequestInterceptor {
+ /**
+ * Intercepts handling of an incoming request. If non-null message is returned, the standard handling is omitted.
+ *
+ * @param service Server service.
+ * @param request Request in question.
+ * @param done Done closure.
+ * @return A message to return to the caller, or {@code null} if standard handling should be used.
+ */
+ @Nullable
+ Message intercept(RaftServerService service, AppendEntriesRequest request, RpcRequestClosure done);
Review Comment:
```suggestion
@Nullable Message intercept(RaftServerService service, AppendEntriesRequest request, RpcRequestClosure done);
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/schemasync/ItSchemaSyncAndReplicationTest.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.schemasync;
+
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.ReplicationGroupsUtils;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.metastorage.server.raft.MetastorageGroupId;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.schema.CheckCatalogVersionOnAppendEntries;
+import org.apache.ignite.internal.test.WatchListenerInhibitor;
+import org.apache.ignite.internal.testframework.jul.NoOpHandler;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests about interaction between Schema Synchronization and Replication.
+ */
+@SuppressWarnings("resource")
+class ItSchemaSyncAndReplicationTest extends ClusterPerTestIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 3;
+ }
+
+ /**
+ * The replication mechanism must not replicate commands for which schemas are not yet available on the node
+ * to which replication happens (in Raft, it means that followers/learners cannot receive commands that they
+ * cannot execute without waiting for schemas). This method tests this scenario.
+ */
+ @Test
+ void laggingSchemasPreventPartitionDataReplication() throws Exception {
+ createTestTableWith3Replicas();
+
+ final int notInhibitedNodeIndex = 0;
+ transferLeadershipsTo(notInhibitedNodeIndex);
+
+ IgniteImpl nodeToInhibitMetaStorage = cluster.node(1);
+
+ WatchListenerInhibitor listenerInhibitor = WatchListenerInhibitor.metastorageEventsInhibitor(nodeToInhibitMetaStorage);
+ listenerInhibitor.startInhibit();
+
+ try {
+ CountDownLatch rejectionTriggered = rejectionDueToMetadataLagTriggered();
+
+ updateTableSchemaAt(notInhibitedNodeIndex);
+ putToTableAt(notInhibitedNodeIndex);
+
+ assertTrue(rejectionTriggered.await(10, TimeUnit.SECONDS), "Did not see rejections due to lagging metadata");
+
+ assertTrue(solePartitionIsEmpty(nodeToInhibitMetaStorage), "Something was written to the partition");
+
+ listenerInhibitor.stopInhibit();
+
+ assertTrue(
+ waitForCondition(() -> !solePartitionIsEmpty(nodeToInhibitMetaStorage), 10_000),
+ "Nothing was written to partition even after inhibiting was cancelled"
+ );
+ } finally {
+ listenerInhibitor.stopInhibit();
+ }
+ }
+
+ private void createTestTableWith3Replicas() throws InterruptedException {
+ String zoneSql = "create zone test_zone with partitions=1, replicas=3";
+ String sql = "create table test (key int primary key, value varchar(20))"
+ + " with primary_zone='TEST_ZONE'";
+
+ cluster.doInSession(0, session -> {
+ executeUpdate(zoneSql, session);
+ executeUpdate(sql, session);
+ });
+
+ waitForTableToStart();
+ }
+
+ private void waitForTableToStart() throws InterruptedException {
+ // TODO: IGNITE-18733 - remove this wait because when a table creation query is executed, the table must be fully ready.
+
+ BooleanSupplier tableStarted = () -> {
+ int numberOfStartedRaftNodes = cluster.runningNodes()
+ .map(ReplicationGroupsUtils::tablePartitionIds)
+ .mapToInt(List::size)
+ .sum();
+ return numberOfStartedRaftNodes == 3;
+ };
+
+ assertTrue(waitForCondition(tableStarted, 10_000), "Did not see all table RAFT nodes started");
+ }
+
+ private void transferLeadershipsTo(int nodeIndex) throws InterruptedException {
+ cluster.transferLeadershipTo(nodeIndex, MetastorageGroupId.INSTANCE);
+ cluster.transferLeadershipTo(nodeIndex, cluster.solePartitionId());
+ }
+
+ private static CountDownLatch rejectionDueToMetadataLagTriggered() {
+ Logger interceptorLogger = Logger.getLogger(CheckCatalogVersionOnAppendEntries.class.getName());
+
+ CountDownLatch rejectionTriggered = new CountDownLatch(1);
Review Comment:
Maybe CompletableFuture?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/CatalogVersionAware.java:
##########
@@ -0,0 +1,28 @@
+/*
+ * 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.command;
+
+/**
+ * A command that requires certain level of catalog version to be locally available just to be accepted on the node.
+ */
+public interface CatalogVersionAware {
Review Comment:
Please add `@FunctionalInterface`
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/ThreadLocalPartitionCommandsMarshaller.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+
+/**
+ * Thread-safe variant of {@link PartitionCommandsMarshaller}.
Review Comment:
`Thread-local` i guess =)
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1158,25 +1169,31 @@ private CompletableFuture<Object> finishTransaction(List<TablePartitionId> aggre
HybridTimestamp currentTimestamp = hybridClock.now();
HybridTimestamp commitTimestamp = commit ? currentTimestamp : null;
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(currentTimestamp.longValue())
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
+ return catalogVersionFor(currentTimestamp)
+ .thenApply(catalogVersion -> {
Review Comment:
Good comment from Vanya, of the same opinion.
##########
modules/core/src/main/java/org/apache/ignite/internal/util/VarIntUtils.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.util;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Utilities to work with general-purpose varints.
Review Comment:
I think we should reveal what "**varints**" is.
And in what format will it be presented.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/schemasync/ItSchemaSyncAndReplicationTest.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.schemasync;
+
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.ReplicationGroupsUtils;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.metastorage.server.raft.MetastorageGroupId;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.schema.CheckCatalogVersionOnAppendEntries;
+import org.apache.ignite.internal.test.WatchListenerInhibitor;
+import org.apache.ignite.internal.testframework.jul.NoOpHandler;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests about interaction between Schema Synchronization and Replication.
+ */
+@SuppressWarnings("resource")
+class ItSchemaSyncAndReplicationTest extends ClusterPerTestIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 3;
+ }
+
+ /**
+ * The replication mechanism must not replicate commands for which schemas are not yet available on the node
+ * to which replication happens (in Raft, it means that followers/learners cannot receive commands that they
+ * cannot execute without waiting for schemas). This method tests this scenario.
+ */
+ @Test
+ void laggingSchemasPreventPartitionDataReplication() throws Exception {
+ createTestTableWith3Replicas();
+
+ final int notInhibitedNodeIndex = 0;
+ transferLeadershipsTo(notInhibitedNodeIndex);
+
+ IgniteImpl nodeToInhibitMetaStorage = cluster.node(1);
+
+ WatchListenerInhibitor listenerInhibitor = WatchListenerInhibitor.metastorageEventsInhibitor(nodeToInhibitMetaStorage);
+ listenerInhibitor.startInhibit();
+
+ try {
+ CountDownLatch rejectionTriggered = rejectionDueToMetadataLagTriggered();
+
+ updateTableSchemaAt(notInhibitedNodeIndex);
+ putToTableAt(notInhibitedNodeIndex);
+
+ assertTrue(rejectionTriggered.await(10, TimeUnit.SECONDS), "Did not see rejections due to lagging metadata");
+
+ assertTrue(solePartitionIsEmpty(nodeToInhibitMetaStorage), "Something was written to the partition");
+
+ listenerInhibitor.stopInhibit();
+
+ assertTrue(
+ waitForCondition(() -> !solePartitionIsEmpty(nodeToInhibitMetaStorage), 10_000),
+ "Nothing was written to partition even after inhibiting was cancelled"
+ );
+ } finally {
+ listenerInhibitor.stopInhibit();
+ }
+ }
+
+ private void createTestTableWith3Replicas() throws InterruptedException {
+ String zoneSql = "create zone test_zone with partitions=1, replicas=3";
+ String sql = "create table test (key int primary key, value varchar(20))"
+ + " with primary_zone='TEST_ZONE'";
+
+ cluster.doInSession(0, session -> {
+ executeUpdate(zoneSql, session);
+ executeUpdate(sql, session);
+ });
+
+ waitForTableToStart();
+ }
+
+ private void waitForTableToStart() throws InterruptedException {
+ // TODO: IGNITE-18733 - remove this wait because when a table creation query is executed, the table must be fully ready.
+
+ BooleanSupplier tableStarted = () -> {
+ int numberOfStartedRaftNodes = cluster.runningNodes()
+ .map(ReplicationGroupsUtils::tablePartitionIds)
+ .mapToInt(List::size)
+ .sum();
+ return numberOfStartedRaftNodes == 3;
Review Comment:
Maybe instead 3 use constant or method argument?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/option/NodeOptions.java:
##########
@@ -704,4 +705,16 @@ public TimeoutStrategy getElectionTimeoutStrategy() {
public void setElectionTimeoutStrategy(TimeoutStrategy electionTimeoutStrategy) {
this.electionTimeoutStrategy = electionTimeoutStrategy;
}
+
+ public Marshaller getCommandsMarshaller() {
Review Comment:
Nullable?
##########
modules/core/src/test/java/org/apache/ignite/internal/util/VarIntUtilsTest.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.util;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import java.nio.ByteBuffer;
+import java.util.stream.Stream;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class VarIntUtilsTest {
+ private final byte[] array = new byte[10];
Review Comment:
I suggest using it as a local variable, it's easier to read the code.
##########
modules/runner/src/testFixtures/java/org/apache/ignite/internal/Cluster.java:
##########
@@ -523,6 +531,74 @@ public void removeNetworkPartitionOf(int nodeIndex) {
LOG.info("Reanimated node " + nodeIndex + " by removing an artificial network partition");
}
+ /**
+ * Transfers leadsership over a replication group to a node identified by the given index.
+ *
+ * @param nodeIndex Node index of the new leader.
+ * @param groupId ID of the replication group.
+ * @throws InterruptedException If interrupted while waiting.
+ */
+ public void transferLeadershipTo(int nodeIndex, ReplicationGroupId groupId) throws InterruptedException {
+ String nodeConsistentId = node(nodeIndex).node().name();
+
+ int maxAttempts = 3;
+
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ boolean transferred = tryTransferLeadershipTo(nodeConsistentId, groupId);
+
+ if (transferred) {
+ break;
+ }
+
+ if (attempt < maxAttempts) {
+ LOG.info("Did not transfer leadership after " + attempt + " attempts, going to retry...");
+ } else {
+ fail("Did not transfer leadership in time after " + maxAttempts + " attempts");
Review Comment:
```suggestion
fail("Did not transfer leadership in time after {} attempts", maxAttempts);
```
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -122,10 +125,12 @@ public class JraftServerImpl implements RaftServer {
private ExecutorService requestExecutor;
/** Marshaller for RAFT commands. */
- private final Marshaller commandsMarshaller;
+ private final Marshaller defaultCommandsMarshaller;
/** Raft service event interceptor. */
- private RaftServiceEventInterceptor serviceEventInterceptor;
+ private final RaftServiceEventInterceptor serviceEventInterceptor;
+
+ private AppendEntriesRequestInterceptor appendEntriesRequestInterceptor = new NullAppendEntriesRequestInterceptor();
Review Comment:
Needs `volatile`?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/schemasync/ItSchemaSyncAndReplicationTest.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.schemasync;
+
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.ReplicationGroupsUtils;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.metastorage.server.raft.MetastorageGroupId;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.schema.CheckCatalogVersionOnAppendEntries;
+import org.apache.ignite.internal.test.WatchListenerInhibitor;
+import org.apache.ignite.internal.testframework.jul.NoOpHandler;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests about interaction between Schema Synchronization and Replication.
+ */
+@SuppressWarnings("resource")
+class ItSchemaSyncAndReplicationTest extends ClusterPerTestIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 3;
+ }
+
+ /**
+ * The replication mechanism must not replicate commands for which schemas are not yet available on the node
+ * to which replication happens (in Raft, it means that followers/learners cannot receive commands that they
+ * cannot execute without waiting for schemas). This method tests this scenario.
+ */
+ @Test
+ void laggingSchemasPreventPartitionDataReplication() throws Exception {
+ createTestTableWith3Replicas();
+
+ final int notInhibitedNodeIndex = 0;
+ transferLeadershipsTo(notInhibitedNodeIndex);
+
+ IgniteImpl nodeToInhibitMetaStorage = cluster.node(1);
+
+ WatchListenerInhibitor listenerInhibitor = WatchListenerInhibitor.metastorageEventsInhibitor(nodeToInhibitMetaStorage);
+ listenerInhibitor.startInhibit();
+
+ try {
+ CountDownLatch rejectionTriggered = rejectionDueToMetadataLagTriggered();
+
+ updateTableSchemaAt(notInhibitedNodeIndex);
+ putToTableAt(notInhibitedNodeIndex);
+
+ assertTrue(rejectionTriggered.await(10, TimeUnit.SECONDS), "Did not see rejections due to lagging metadata");
+
+ assertTrue(solePartitionIsEmpty(nodeToInhibitMetaStorage), "Something was written to the partition");
+
+ listenerInhibitor.stopInhibit();
+
+ assertTrue(
+ waitForCondition(() -> !solePartitionIsEmpty(nodeToInhibitMetaStorage), 10_000),
+ "Nothing was written to partition even after inhibiting was cancelled"
+ );
+ } finally {
+ listenerInhibitor.stopInhibit();
+ }
+ }
+
+ private void createTestTableWith3Replicas() throws InterruptedException {
+ String zoneSql = "create zone test_zone with partitions=1, replicas=3";
+ String sql = "create table test (key int primary key, value varchar(20))"
+ + " with primary_zone='TEST_ZONE'";
+
+ cluster.doInSession(0, session -> {
+ executeUpdate(zoneSql, session);
+ executeUpdate(sql, session);
+ });
+
+ waitForTableToStart();
+ }
+
+ private void waitForTableToStart() throws InterruptedException {
+ // TODO: IGNITE-18733 - remove this wait because when a table creation query is executed, the table must be fully ready.
+
+ BooleanSupplier tableStarted = () -> {
+ int numberOfStartedRaftNodes = cluster.runningNodes()
+ .map(ReplicationGroupsUtils::tablePartitionIds)
+ .mapToInt(List::size)
+ .sum();
+ return numberOfStartedRaftNodes == 3;
+ };
+
+ assertTrue(waitForCondition(tableStarted, 10_000), "Did not see all table RAFT nodes started");
+ }
+
+ private void transferLeadershipsTo(int nodeIndex) throws InterruptedException {
+ cluster.transferLeadershipTo(nodeIndex, MetastorageGroupId.INSTANCE);
+ cluster.transferLeadershipTo(nodeIndex, cluster.solePartitionId());
+ }
+
+ private static CountDownLatch rejectionDueToMetadataLagTriggered() {
+ Logger interceptorLogger = Logger.getLogger(CheckCatalogVersionOnAppendEntries.class.getName());
+
+ CountDownLatch rejectionTriggered = new CountDownLatch(1);
+
+ interceptorLogger.addHandler(new NoOpHandler() {
+ @Override
+ public void publish(LogRecord record) {
+ if (record.getMessage().startsWith("Metadata not yet available")) {
+ rejectionTriggered.countDown();
+ }
+ }
+ });
+
+ return rejectionTriggered;
+ }
+
+ private void putToTableAt(int nodeIndex) {
+ KeyValueView<Tuple, Tuple> kvView = cluster.node(nodeIndex)
+ .tables()
+ .table("test")
+ .keyValueView();
+ kvView.put(null, Tuple.create().set("key", 1), Tuple.create().set("value", "one"));
Review Comment:
```suggestion
cluster.node(nodeIndex)
.tables()
.table("test")
.keyValueView()
.put(null, Tuple.create().set("key", 1), Tuple.create().set("value", "one"));
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/PartitionCommandsMarshallerImpl.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.raft.util.OptimizedMarshaller;
+import org.apache.ignite.internal.table.distributed.command.CatalogVersionAware;
+import org.apache.ignite.internal.util.VarIntUtils;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+
+/**
+ * Default {@link PartitionCommandsMarshaller} implementation.
+ */
+public class PartitionCommandsMarshallerImpl extends OptimizedMarshaller implements PartitionCommandsMarshaller {
+ public PartitionCommandsMarshallerImpl(MessageSerializationRegistry serializationRegistry) {
+ super(serializationRegistry);
+ }
+
+ @Override
+ public byte[] marshall(Object o) {
+ int requiredCatalogVersion = o instanceof CatalogVersionAware ? ((CatalogVersionAware) o).requiredCatalogVersion() : -1;
Review Comment:
did not notice where it is said about `-1` and its meaning.
##########
modules/runner/src/testFixtures/java/org/apache/ignite/internal/Cluster.java:
##########
@@ -523,6 +531,74 @@ public void removeNetworkPartitionOf(int nodeIndex) {
LOG.info("Reanimated node " + nodeIndex + " by removing an artificial network partition");
}
+ /**
+ * Transfers leadsership over a replication group to a node identified by the given index.
+ *
+ * @param nodeIndex Node index of the new leader.
+ * @param groupId ID of the replication group.
+ * @throws InterruptedException If interrupted while waiting.
+ */
+ public void transferLeadershipTo(int nodeIndex, ReplicationGroupId groupId) throws InterruptedException {
+ String nodeConsistentId = node(nodeIndex).node().name();
+
+ int maxAttempts = 3;
+
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ boolean transferred = tryTransferLeadershipTo(nodeConsistentId, groupId);
+
+ if (transferred) {
+ break;
+ }
+
+ if (attempt < maxAttempts) {
+ LOG.info("Did not transfer leadership after " + attempt + " attempts, going to retry...");
Review Comment:
```suggestion
LOG.info("Did not transfer leadership after {} attempts, going to retry...", attempt);
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/schemasync/ItSchemaSyncAndReplicationTest.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.schemasync;
+
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.ReplicationGroupsUtils;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.metastorage.server.raft.MetastorageGroupId;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.schema.CheckCatalogVersionOnAppendEntries;
+import org.apache.ignite.internal.test.WatchListenerInhibitor;
+import org.apache.ignite.internal.testframework.jul.NoOpHandler;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests about interaction between Schema Synchronization and Replication.
+ */
+@SuppressWarnings("resource")
+class ItSchemaSyncAndReplicationTest extends ClusterPerTestIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 3;
+ }
+
+ /**
+ * The replication mechanism must not replicate commands for which schemas are not yet available on the node
+ * to which replication happens (in Raft, it means that followers/learners cannot receive commands that they
+ * cannot execute without waiting for schemas). This method tests this scenario.
+ */
+ @Test
+ void laggingSchemasPreventPartitionDataReplication() throws Exception {
+ createTestTableWith3Replicas();
+
+ final int notInhibitedNodeIndex = 0;
+ transferLeadershipsTo(notInhibitedNodeIndex);
+
+ IgniteImpl nodeToInhibitMetaStorage = cluster.node(1);
+
+ WatchListenerInhibitor listenerInhibitor = WatchListenerInhibitor.metastorageEventsInhibitor(nodeToInhibitMetaStorage);
+ listenerInhibitor.startInhibit();
+
+ try {
+ CountDownLatch rejectionTriggered = rejectionDueToMetadataLagTriggered();
+
+ updateTableSchemaAt(notInhibitedNodeIndex);
+ putToTableAt(notInhibitedNodeIndex);
+
+ assertTrue(rejectionTriggered.await(10, TimeUnit.SECONDS), "Did not see rejections due to lagging metadata");
+
+ assertTrue(solePartitionIsEmpty(nodeToInhibitMetaStorage), "Something was written to the partition");
+
+ listenerInhibitor.stopInhibit();
+
+ assertTrue(
+ waitForCondition(() -> !solePartitionIsEmpty(nodeToInhibitMetaStorage), 10_000),
+ "Nothing was written to partition even after inhibiting was cancelled"
+ );
+ } finally {
+ listenerInhibitor.stopInhibit();
+ }
+ }
+
+ private void createTestTableWith3Replicas() throws InterruptedException {
+ String zoneSql = "create zone test_zone with partitions=1, replicas=3";
+ String sql = "create table test (key int primary key, value varchar(20))"
+ + " with primary_zone='TEST_ZONE'";
+
+ cluster.doInSession(0, session -> {
+ executeUpdate(zoneSql, session);
+ executeUpdate(sql, session);
+ });
+
+ waitForTableToStart();
+ }
+
+ private void waitForTableToStart() throws InterruptedException {
+ // TODO: IGNITE-18733 - remove this wait because when a table creation query is executed, the table must be fully ready.
+
+ BooleanSupplier tableStarted = () -> {
+ int numberOfStartedRaftNodes = cluster.runningNodes()
+ .map(ReplicationGroupsUtils::tablePartitionIds)
+ .mapToInt(List::size)
+ .sum();
+ return numberOfStartedRaftNodes == 3;
+ };
+
+ assertTrue(waitForCondition(tableStarted, 10_000), "Did not see all table RAFT nodes started");
+ }
+
+ private void transferLeadershipsTo(int nodeIndex) throws InterruptedException {
+ cluster.transferLeadershipTo(nodeIndex, MetastorageGroupId.INSTANCE);
+ cluster.transferLeadershipTo(nodeIndex, cluster.solePartitionId());
+ }
+
+ private static CountDownLatch rejectionDueToMetadataLagTriggered() {
+ Logger interceptorLogger = Logger.getLogger(CheckCatalogVersionOnAppendEntries.class.getName());
+
+ CountDownLatch rejectionTriggered = new CountDownLatch(1);
+
+ interceptorLogger.addHandler(new NoOpHandler() {
+ @Override
+ public void publish(LogRecord record) {
+ if (record.getMessage().startsWith("Metadata not yet available")) {
+ rejectionTriggered.countDown();
+ }
+ }
+ });
+
+ return rejectionTriggered;
+ }
+
+ private void putToTableAt(int nodeIndex) {
+ KeyValueView<Tuple, Tuple> kvView = cluster.node(nodeIndex)
+ .tables()
+ .table("test")
+ .keyValueView();
+ kvView.put(null, Tuple.create().set("key", 1), Tuple.create().set("value", "one"));
+ }
+
+ private void updateTableSchemaAt(int nodeIndex) {
+ cluster.doInSession(nodeIndex, session -> {
+ session.execute(null, "alter table test add column added int");
+ });
+ }
+
+ private static boolean solePartitionIsEmpty(IgniteImpl node) {
+ MvPartitionStorage mvPartitionStorage = solePartitionStorage(node);
+ RowId rowId = mvPartitionStorage.closestRowId(RowId.lowestRowId(0));
+ return rowId == null;
+ }
+
+ private static MvPartitionStorage solePartitionStorage(IgniteImpl node) {
+ TableImpl table = (TableImpl) node.tables().table("test");
Review Comment:
Let's move "table" to constant.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnAppendEntries.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.raft.jraft.Node;
+import org.apache.ignite.raft.jraft.entity.EnumOutter.EntryType;
+import org.apache.ignite.raft.jraft.entity.RaftOutter;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.EntryMeta;
+import org.apache.ignite.raft.jraft.error.RaftError;
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.rpc.RaftRpcFactory;
+import org.apache.ignite.raft.jraft.rpc.RaftServerService;
+import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
+import org.apache.ignite.raft.jraft.rpc.impl.AppendEntriesRequestInterceptor;
+import org.apache.ignite.raft.jraft.util.Marshaller;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An {@link AppendEntriesRequestInterceptor} that rejects requests (by returning EBUSY error code) if any of the
+ * incoming commands requires catalog version that is not available locally yet.
+ */
+public class CheckCatalogVersionOnAppendEntries implements AppendEntriesRequestInterceptor {
+ private static final IgniteLogger LOG = Loggers.forClass(CheckCatalogVersionOnAppendEntries.class);
+
+ private static final int NO_VERSION_REQUIREMENT = Integer.MIN_VALUE;
+
+ private final CatalogService catalogService;
+
+ public CheckCatalogVersionOnAppendEntries(CatalogService catalogService) {
+ this.catalogService = catalogService;
+ }
+
+ @Override
+ @Nullable
+ public Message intercept(RaftServerService service, AppendEntriesRequest request, RpcRequestClosure done) {
+ if (request.entriesList() == null || request.data() == null) {
+ return null;
+ }
+
+ Node node = (Node) service;
+
+ ByteBuffer allData = request.data().asReadOnlyByteBuffer();
+ int offset = 0;
+
+ for (RaftOutter.EntryMeta entry : request.entriesList()) {
+ int requiredCatalogVersion = readRequiredCatalogVersionForMeta(allData, entry, node.getOptions().requiredCommandsMarshaller());
+
+ if (requiredCatalogVersion != NO_VERSION_REQUIREMENT && !isMetadataAvailableFor(requiredCatalogVersion)) {
+ LOG.warn("Metadata not yet available, group {}, required level {}.", request.groupId(), requiredCatalogVersion);
+ return RaftRpcFactory.DEFAULT //
+ .newResponse(node.getRaftOptions().getRaftMessagesFactory(), RaftError.EBUSY,
+ "Metadata not yet available, group '%s', required level %d.", request.groupId(), requiredCatalogVersion);
+ }
+
+ offset += (int) entry.dataLen();
+ allData.position(offset);
+ }
+
+ return null;
+ }
+
+ private static int readRequiredCatalogVersionForMeta(ByteBuffer allData, final EntryMeta entry, Marshaller commandsMarshaller) {
+ if (entry.type() != EntryType.ENTRY_TYPE_DATA) {
+ return NO_VERSION_REQUIREMENT;
+ }
+
+ if (!(commandsMarshaller instanceof PartitionCommandsMarshaller)) {
+ return NO_VERSION_REQUIREMENT;
+ }
+
+ PartitionCommandsMarshaller partitionCommandsMarshaller = (PartitionCommandsMarshaller) commandsMarshaller;
+
+ long dataLen = entry.dataLen();
+ if (dataLen > 0) {
+ return partitionCommandsMarshaller.readRequiredCatalogVersion(allData);
+ }
+
+ return NO_VERSION_REQUIREMENT;
+ }
+
+ private boolean isMetadataAvailableFor(int catalogVersion) {
Review Comment:
Maybe `catalogVersion` -> `requiredCatalogVersion` ?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/PartitionCommandsMarshaller.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.raft.jraft.util.Marshaller;
+
+/**
+ * {@link Marshaller} that first writes some metadata about an object and then it writes the actual serialized
+ * representation of the object.
+ */
+public interface PartitionCommandsMarshaller extends Marshaller {
Review Comment:
Maybe add `@FunctionalInterface` ?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/AppendEntriesRequestInterceptor.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.raft.jraft.rpc.impl;
+
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.rpc.RaftServerService;
+import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Intercepts {@link AppendEntriesRequest}s as they come in. It might be used to handle such a request in a non-standard
+ * way (like returning EBUSY under special circumstances instead of the standard behavior).
+ */
+public interface AppendEntriesRequestInterceptor {
Review Comment:
Maybe add `@FunctionalInterface`
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java:
##########
@@ -1590,6 +1602,82 @@ public void failsWhenFullScanReadsTupleWithIncompatibleSchemaFromFuture() {
);
}
+ @ParameterizedTest
+ @MethodSource("singleRowWriteRequestTypes")
+ public void singleRowWritesAreSuppliedWithRequiredCatalogVersion(RequestType requestType) {
+ testWritesAreSuppliedWithRequiredCatalogVersion(requestType, (targetTxId, key) -> {
+ return doSingleRowRequest(targetTxId, marshalKeyOrKeyValue(requestType, key), requestType);
+ });
+ }
+
+ private static Stream<Arguments> singleRowWriteRequestTypes() {
+ return Arrays.stream(RequestType.values())
+ .filter(RequestType::isSingleRowWrite)
+ .map(Arguments::of);
+ }
+
+ private void testWritesAreSuppliedWithRequiredCatalogVersion(RequestType requestType, ListenerInvocation listenerInvocation) {
+ TestKey key = nextKey();
+
+ if (requestType.looksUpFirst()) {
+ UUID tx0 = beginTx();
+ upsert(tx0, binaryRow(key, someValue));
+ cleanup(tx0);
+
+ // While handling the upsert, our mocks were touched, let's reset them to prevent false-positives during verification.
+ Mockito.reset(schemaSyncService);
+ }
+
+ when(catalogService.activeCatalogVersion(anyLong())).thenReturn(42);
+
+ UUID targetTxId = beginTx();
+
+ CompletableFuture<?> future = listenerInvocation.invoke(targetTxId, key);
+
+ assertThat(future, willCompleteSuccessfully());
+
+ // Make sure metadata completeness is awaited for.
+ InOrder inOrder = inOrder(schemaSyncService, catalogService);
+
Review Comment:
```suggestion
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/action/RequestType.java:
##########
@@ -79,6 +79,13 @@ public boolean isSingleRow() {
}
}
+ /**
+ * Returns {@code true} if the operation works with a single row and it's a write.
+ */
+ public boolean isSingleRowWrite() {
Review Comment:
Maybe add a test utility class for this in `testFixtures`?
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java:
##########
@@ -1423,30 +1451,22 @@ private static List<FullTableSchema> incompatibleSchemaVersions(int fromSchemaVe
@MethodSource("singleRowRequestTypes")
public void failsWhenReadingSingleRowFromFutureIncompatibleSchema(RequestType requestType) {
testFailsWhenReadingFromFutureIncompatibleSchema((targetTxId, key) -> {
- try {
- switch (requestType) {
- case RW_GET:
- case RW_DELETE:
- case RW_GET_AND_DELETE:
- return doSingleRowRequest(targetTxId, kvMarshaller.marshal(key), requestType);
-
- case RW_DELETE_EXACT:
- case RW_INSERT:
- case RW_UPSERT:
- case RW_GET_AND_UPSERT:
- case RW_GET_AND_REPLACE:
- case RW_REPLACE_IF_EXIST:
- return doSingleRowRequest(targetTxId, kvMarshaller.marshal(key, new TestValue(1, "v1")), requestType);
-
- default:
- throw new AssertionError("Unexpected operation type: " + requestType);
- }
- } catch (MarshallerException e) {
- throw new AssertionError(e);
- }
+ return doSingleRowRequest(targetTxId, marshalKeyOrKeyValue(requestType, key), requestType);
});
}
+ private BinaryRow marshalKeyOrKeyValue(RequestType requestType, TestKey key) {
+ try {
+ if (requestType.isKeyOnly()) {
+ return kvMarshaller.marshal(key);
+ } else {
+ return kvMarshaller.marshal(key, someValue);
+ }
Review Comment:
```suggestion
return requestType.isKeyOnly() ? kvMarshaller.marshal(key) : kvMarshaller.marshal(key, someValue);
```
--
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] tkalkirill commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1302491260
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnAppendEntries.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.raft.jraft.Node;
+import org.apache.ignite.raft.jraft.entity.EnumOutter.EntryType;
+import org.apache.ignite.raft.jraft.entity.RaftOutter;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.EntryMeta;
+import org.apache.ignite.raft.jraft.error.RaftError;
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.rpc.RaftRpcFactory;
+import org.apache.ignite.raft.jraft.rpc.RaftServerService;
+import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
+import org.apache.ignite.raft.jraft.rpc.impl.AppendEntriesRequestInterceptor;
+import org.apache.ignite.raft.jraft.util.Marshaller;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An {@link AppendEntriesRequestInterceptor} that rejects requests (by returning EBUSY error code) if any of the
+ * incoming commands requires catalog version that is not available locally yet.
+ */
+public class CheckCatalogVersionOnAppendEntries implements AppendEntriesRequestInterceptor {
+ private static final IgniteLogger LOG = Loggers.forClass(CheckCatalogVersionOnAppendEntries.class);
+
+ private static final int NO_VERSION_REQUIREMENT = Integer.MIN_VALUE;
+
+ private final CatalogService catalogService;
+
+ public CheckCatalogVersionOnAppendEntries(CatalogService catalogService) {
+ this.catalogService = catalogService;
+ }
+
+ @Override
+ @Nullable
+ public Message intercept(RaftServerService service, AppendEntriesRequest request, RpcRequestClosure done) {
Review Comment:
```suggestion
public @Nullable Message intercept(RaftServerService service, AppendEntriesRequest request, RpcRequestClosure done) {
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/schemasync/ItSchemaSyncAndReplicationTest.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.schemasync;
+
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.ReplicationGroupsUtils;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.metastorage.server.raft.MetastorageGroupId;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.schema.CheckCatalogVersionOnAppendEntries;
+import org.apache.ignite.internal.test.WatchListenerInhibitor;
+import org.apache.ignite.internal.testframework.jul.NoOpHandler;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests about interaction between Schema Synchronization and Replication.
+ */
+@SuppressWarnings("resource")
+class ItSchemaSyncAndReplicationTest extends ClusterPerTestIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 3;
+ }
+
+ /**
+ * The replication mechanism must not replicate commands for which schemas are not yet available on the node
+ * to which replication happens (in Raft, it means that followers/learners cannot receive commands that they
+ * cannot execute without waiting for schemas). This method tests this scenario.
+ */
+ @Test
+ void laggingSchemasPreventPartitionDataReplication() throws Exception {
+ createTestTableWith3Replicas();
+
+ final int notInhibitedNodeIndex = 0;
+ transferLeadershipsTo(notInhibitedNodeIndex);
+
+ IgniteImpl nodeToInhibitMetaStorage = cluster.node(1);
+
+ WatchListenerInhibitor listenerInhibitor = WatchListenerInhibitor.metastorageEventsInhibitor(nodeToInhibitMetaStorage);
+ listenerInhibitor.startInhibit();
+
+ try {
+ CountDownLatch rejectionTriggered = rejectionDueToMetadataLagTriggered();
+
+ updateTableSchemaAt(notInhibitedNodeIndex);
+ putToTableAt(notInhibitedNodeIndex);
+
+ assertTrue(rejectionTriggered.await(10, TimeUnit.SECONDS), "Did not see rejections due to lagging metadata");
+
+ assertTrue(solePartitionIsEmpty(nodeToInhibitMetaStorage), "Something was written to the partition");
+
+ listenerInhibitor.stopInhibit();
+
+ assertTrue(
+ waitForCondition(() -> !solePartitionIsEmpty(nodeToInhibitMetaStorage), 10_000),
+ "Nothing was written to partition even after inhibiting was cancelled"
+ );
+ } finally {
+ listenerInhibitor.stopInhibit();
+ }
+ }
+
+ private void createTestTableWith3Replicas() throws InterruptedException {
+ String zoneSql = "create zone test_zone with partitions=1, replicas=3";
+ String sql = "create table test (key int primary key, value varchar(20))"
+ + " with primary_zone='TEST_ZONE'";
+
+ cluster.doInSession(0, session -> {
+ executeUpdate(zoneSql, session);
+ executeUpdate(sql, session);
+ });
+
+ waitForTableToStart();
+ }
+
+ private void waitForTableToStart() throws InterruptedException {
+ // TODO: IGNITE-18733 - remove this wait because when a table creation query is executed, the table must be fully ready.
+
+ BooleanSupplier tableStarted = () -> {
+ int numberOfStartedRaftNodes = cluster.runningNodes()
+ .map(ReplicationGroupsUtils::tablePartitionIds)
+ .mapToInt(List::size)
+ .sum();
+ return numberOfStartedRaftNodes == 3;
+ };
+
+ assertTrue(waitForCondition(tableStarted, 10_000), "Did not see all table RAFT nodes started");
+ }
+
+ private void transferLeadershipsTo(int nodeIndex) throws InterruptedException {
+ cluster.transferLeadershipTo(nodeIndex, MetastorageGroupId.INSTANCE);
+ cluster.transferLeadershipTo(nodeIndex, cluster.solePartitionId());
+ }
+
+ private static CountDownLatch rejectionDueToMetadataLagTriggered() {
+ Logger interceptorLogger = Logger.getLogger(CheckCatalogVersionOnAppendEntries.class.getName());
+
+ CountDownLatch rejectionTriggered = new CountDownLatch(1);
Review Comment:
Ease of use, instead of `assertTrue(rejectionTriggered.await(10, TimeUnit.SECONDS), "Did not see rejections due to lagging metadata");` you can use `assertThat(future, willCompleteSuccessfully());`, up to u.
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -122,10 +125,12 @@ public class JraftServerImpl implements RaftServer {
private ExecutorService requestExecutor;
/** Marshaller for RAFT commands. */
- private final Marshaller commandsMarshaller;
+ private final Marshaller defaultCommandsMarshaller;
/** Raft service event interceptor. */
- private RaftServiceEventInterceptor serviceEventInterceptor;
+ private final RaftServiceEventInterceptor serviceEventInterceptor;
+
+ private AppendEntriesRequestInterceptor appendEntriesRequestInterceptor = new NullAppendEntriesRequestInterceptor();
Review Comment:
Let's point it out that it's not thread safe.
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -122,10 +125,12 @@ public class JraftServerImpl implements RaftServer {
private ExecutorService requestExecutor;
/** Marshaller for RAFT commands. */
- private final Marshaller commandsMarshaller;
+ private final Marshaller defaultCommandsMarshaller;
Review Comment:
Please fix it.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1158,25 +1169,31 @@ private CompletableFuture<Object> finishTransaction(List<TablePartitionId> aggre
HybridTimestamp currentTimestamp = hybridClock.now();
HybridTimestamp commitTimestamp = commit ? currentTimestamp : null;
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(currentTimestamp.longValue())
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
+ return catalogVersionFor(currentTimestamp)
+ .thenApply(catalogVersion -> {
Review Comment:
wait
##########
modules/runner/src/testFixtures/java/org/apache/ignite/internal/Cluster.java:
##########
@@ -523,6 +531,74 @@ public void removeNetworkPartitionOf(int nodeIndex) {
LOG.info("Reanimated node " + nodeIndex + " by removing an artificial network partition");
}
+ /**
+ * Transfers leadsership over a replication group to a node identified by the given index.
+ *
+ * @param nodeIndex Node index of the new leader.
+ * @param groupId ID of the replication group.
+ * @throws InterruptedException If interrupted while waiting.
+ */
+ public void transferLeadershipTo(int nodeIndex, ReplicationGroupId groupId) throws InterruptedException {
+ String nodeConsistentId = node(nodeIndex).node().name();
+
+ int maxAttempts = 3;
+
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ boolean transferred = tryTransferLeadershipTo(nodeConsistentId, groupId);
+
+ if (transferred) {
+ break;
+ }
+
+ if (attempt < maxAttempts) {
+ LOG.info("Did not transfer leadership after " + attempt + " attempts, going to retry...");
+ } else {
+ fail("Did not transfer leadership in time after " + maxAttempts + " attempts");
Review Comment:
=)
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1307555068
##########
modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java:
##########
@@ -513,12 +517,18 @@ public class IgniteImpl implements Ignite {
SchemaSynchronizationConfiguration.KEY
);
+ LongSupplier delayDurationMsSupplier = () -> schemaSyncConfig.delayDuration().value();
+
catalogManager = new CatalogManagerImpl(
new UpdateLogImpl(metaStorageMgr),
clockWaiter,
- () -> schemaSyncConfig.delayDuration().value()
+ delayDurationMsSupplier
Review Comment:
Because at the moment the constructor is called the node is not started yet, and the configuration becomes available only after we start the node
--
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] ibessonov commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1302605781
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java:
##########
@@ -146,6 +147,16 @@ public Loza(
);
}
+ /**
+ * Sets {@link AppendEntriesRequestInterceptor} to use.
+ *
+ * @param appendEntriesRequestInterceptor Interceptor to use.
+ */
+ public void appendEntriesRequestInterceptor(AppendEntriesRequestInterceptor appendEntriesRequestInterceptor) {
Review Comment:
Ok, sounds like a bad design, but we won't be fixing it right here
--
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] tkalkirill merged pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill merged PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301511715
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1158,25 +1169,31 @@ private CompletableFuture<Object> finishTransaction(List<TablePartitionId> aggre
HybridTimestamp currentTimestamp = hybridClock.now();
HybridTimestamp commitTimestamp = commit ? currentTimestamp : null;
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(currentTimestamp.longValue())
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
+ return catalogVersionFor(currentTimestamp)
+ .thenApply(catalogVersion -> {
Review Comment:
According to https://issues.apache.org/jira/browse/IGNITE-20116 , linearizibility is already broken/not guaranteed. Let's wait for @sanpwc to decide what to do here, but probably this should not be fixed in this PR.
--
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] tkalkirill commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301599414
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -122,10 +125,12 @@ public class JraftServerImpl implements RaftServer {
private ExecutorService requestExecutor;
/** Marshaller for RAFT commands. */
- private final Marshaller commandsMarshaller;
+ private final Marshaller defaultCommandsMarshaller;
Review Comment:
JavaDoc doesn't say anything about this.
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301684460
##########
modules/runner/src/testFixtures/java/org/apache/ignite/internal/Cluster.java:
##########
@@ -523,6 +531,74 @@ public void removeNetworkPartitionOf(int nodeIndex) {
LOG.info("Reanimated node " + nodeIndex + " by removing an artificial network partition");
}
+ /**
+ * Transfers leadsership over a replication group to a node identified by the given index.
+ *
+ * @param nodeIndex Node index of the new leader.
+ * @param groupId ID of the replication group.
+ * @throws InterruptedException If interrupted while waiting.
+ */
+ public void transferLeadershipTo(int nodeIndex, ReplicationGroupId groupId) throws InterruptedException {
+ String nodeConsistentId = node(nodeIndex).node().name();
+
+ int maxAttempts = 3;
+
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ boolean transferred = tryTransferLeadershipTo(nodeConsistentId, groupId);
+
+ if (transferred) {
+ break;
+ }
+
+ if (attempt < maxAttempts) {
+ LOG.info("Did not transfer leadership after " + attempt + " attempts, going to retry...");
+ } else {
+ fail("Did not transfer leadership in time after " + maxAttempts + " attempts");
Review Comment:
It's not a logger call :)
--
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] rpuch commented on a diff in pull request #2450: IGNITE-19227 Wait for schema availability outside JRaft threads
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2450:
URL: https://github.com/apache/ignite-3/pull/2450#discussion_r1301704024
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/PartitionCommandsMarshaller.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.raft.jraft.util.Marshaller;
+
+/**
+ * {@link Marshaller} that first writes some metadata about an object and then it writes the actual serialized
+ * representation of the object.
+ */
+public interface PartitionCommandsMarshaller extends Marshaller {
Review Comment:
It cannot be a functional interface as it has 3 methods, not one
--
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