You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ir...@apache.org on 2018/06/18 12:08:26 UTC
ignite git commit: IGNITE-8757 idle_verify utility doesn't show both
update counter and hash conflicts
Repository: ignite
Updated Branches:
refs/heads/master 63e71ea60 -> 9131e4ddb
IGNITE-8757 idle_verify utility doesn't show both update counter and hash conflicts
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/9131e4dd
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/9131e4dd
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/9131e4dd
Branch: refs/heads/master
Commit: 9131e4ddb9ba3cf0faed4d64b4538b96547a9db5
Parents: 63e71ea
Author: Ivan Rakov <ir...@apache.org>
Authored: Mon Jun 18 15:07:48 2018 +0300
Committer: Ivan Rakov <ir...@apache.org>
Committed: Mon Jun 18 15:07:48 2018 +0300
----------------------------------------------------------------------
.../internal/commandline/CommandHandler.java | 95 +++++
.../cache/verify/IdleVerifyResultV2.java | 113 ++++++
.../cache/verify/PartitionHashRecordV2.java | 168 +++++++++
.../processors/cache/verify/PartitionKeyV2.java | 127 +++++++
.../verify/VerifyBackupPartitionsTask.java | 16 +-
.../verify/VerifyBackupPartitionsTaskV2.java | 347 +++++++++++++++++++
.../visor/verify/VisorIdleVerifyTask.java | 2 +
.../visor/verify/VisorIdleVerifyTaskV2.java | 94 +++++
.../resources/META-INF/classnames.properties | 1 +
.../core/src/main/resources/ignite.properties | 2 +-
.../junits/common/GridCommonAbstractTest.java | 6 +-
.../ignite/util/GridCommandHandlerTest.java | 134 +++++++
12 files changed, 1093 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/main/java/org/apache/ignite/internal/commandline/CommandHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/commandline/CommandHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/commandline/CommandHandler.java
index c59e348..be33991 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/commandline/CommandHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/commandline/CommandHandler.java
@@ -34,6 +34,7 @@ import java.util.regex.PatternSyntaxException;
import java.util.stream.Collectors;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.IgniteNodeAttributes;
import org.apache.ignite.internal.client.GridClient;
import org.apache.ignite.internal.client.GridClientAuthenticationException;
import org.apache.ignite.internal.client.GridClientClosedException;
@@ -51,8 +52,12 @@ import org.apache.ignite.internal.commandline.cache.CacheArguments;
import org.apache.ignite.internal.commandline.cache.CacheCommand;
import org.apache.ignite.internal.processors.cache.verify.CacheInfo;
import org.apache.ignite.internal.processors.cache.verify.ContentionInfo;
+import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
+import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecordV2;
import org.apache.ignite.internal.processors.cache.verify.PartitionKey;
+import org.apache.ignite.internal.processors.cache.verify.PartitionKeyV2;
+import org.apache.ignite.internal.processors.cache.verify.VerifyBackupPartitionsTaskV2;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -82,6 +87,7 @@ import org.apache.ignite.internal.visor.verify.VisorContentionTaskResult;
import org.apache.ignite.internal.visor.verify.VisorIdleVerifyTask;
import org.apache.ignite.internal.visor.verify.VisorIdleVerifyTaskArg;
import org.apache.ignite.internal.visor.verify.VisorIdleVerifyTaskResult;
+import org.apache.ignite.internal.visor.verify.VisorIdleVerifyTaskV2;
import org.apache.ignite.internal.visor.verify.VisorValidateIndexesJobResult;
import org.apache.ignite.internal.visor.verify.VisorValidateIndexesTaskArg;
import org.apache.ignite.internal.visor.verify.VisorValidateIndexesTaskResult;
@@ -89,6 +95,7 @@ import org.apache.ignite.internal.visor.verify.VisorViewCacheTask;
import org.apache.ignite.internal.visor.verify.VisorViewCacheTaskArg;
import org.apache.ignite.internal.visor.verify.VisorViewCacheTaskResult;
import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgniteProductVersion;
import org.apache.ignite.plugin.security.SecurityCredentials;
import org.apache.ignite.plugin.security.SecurityCredentialsBasicProvider;
@@ -710,10 +717,39 @@ public class CommandHandler {
}
/**
+ * Executes appropriate version of idle_verify check. Old version will be used if there are old nodes in the cluster.
+ *
* @param client Client.
* @param cacheArgs Cache args.
*/
private void cacheIdleVerify(GridClient client, CacheArguments cacheArgs) throws GridClientException {
+ Collection<GridClientNode> nodes = client.compute().nodes(GridClientNode::connectable);
+
+ boolean idleVerifyV2 = true;
+
+ for (GridClientNode node : nodes) {
+ String nodeVerStr = node.attribute(IgniteNodeAttributes.ATTR_BUILD_VER);
+
+ IgniteProductVersion nodeVer = IgniteProductVersion.fromString(nodeVerStr);
+
+ if (nodeVer.compareTo(VerifyBackupPartitionsTaskV2.V2_SINCE_VER) < 0) {
+ idleVerifyV2 = false;
+
+ break;
+ }
+ }
+
+ if (idleVerifyV2)
+ cacheIdleVerifyV2(client, cacheArgs);
+ else
+ legacyCacheIdleVerify(client, cacheArgs);
+ }
+
+ /**
+ * @param client Client.
+ * @param cacheArgs Cache args.
+ */
+ private void legacyCacheIdleVerify(GridClient client, CacheArguments cacheArgs) throws GridClientException {
VisorIdleVerifyTaskResult res = executeTask(
client, VisorIdleVerifyTask.class, new VisorIdleVerifyTaskArg(cacheArgs.caches()));
@@ -736,6 +772,65 @@ public class CommandHandler {
}
/**
+ * @param client Client.
+ * @param cacheArgs Cache args.
+ */
+ private void cacheIdleVerifyV2(GridClient client, CacheArguments cacheArgs) throws GridClientException {
+ IdleVerifyResultV2 res = executeTask(
+ client, VisorIdleVerifyTaskV2.class, new VisorIdleVerifyTaskArg(cacheArgs.caches()));
+
+ if (!res.hasConflicts()) {
+ log("idle_verify check has finished, no conflicts have been found.");
+ nl();
+ }
+ else {
+ int cntrConflictsSize = res.counterConflicts().size();
+ int hashConflictsSize = res.hashConflicts().size();
+
+ log("idle_verify check has finished, found " + (cntrConflictsSize + hashConflictsSize) +
+ " conflict partitions: [counterConflicts=" + cntrConflictsSize + ", hashConflicts=" +
+ hashConflictsSize + "]");
+ nl();
+
+ if (!F.isEmpty(res.counterConflicts())) {
+ log("Update counter conflicts:");
+
+ for (Map.Entry<PartitionKeyV2, List<PartitionHashRecordV2>> entry : res.counterConflicts().entrySet()) {
+ log("Conflict partition: " + entry.getKey());
+
+ log("Partition instances: " + entry.getValue());
+ }
+
+ nl();
+ }
+
+ if (!F.isEmpty(res.hashConflicts())) {
+ log("Hash conflicts:");
+
+ for (Map.Entry<PartitionKeyV2, List<PartitionHashRecordV2>> entry : res.hashConflicts().entrySet()) {
+ log("Conflict partition: " + entry.getKey());
+
+ log("Partition instances: " + entry.getValue());
+ }
+
+ nl();
+ }
+ }
+
+ if (!F.isEmpty(res.movingPartitions())) {
+ log("Verification was skipped for " + res.movingPartitions().size() + " MOVING partitions:");
+
+ for (Map.Entry<PartitionKeyV2, List<PartitionHashRecordV2>> entry : res.movingPartitions().entrySet()) {
+ log("Rebalancing partition: " + entry.getKey());
+
+ log("Partition instances: " + entry.getValue());
+ }
+
+ nl();
+ }
+ }
+
+ /**
* Change baseline.
*
* @param client Client.
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyResultV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyResultV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyResultV2.java
new file mode 100644
index 0000000..d5815cd
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyResultV2.java
@@ -0,0 +1,113 @@
+/*
+* 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.processors.cache.verify;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.List;
+import java.util.Map;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.internal.visor.VisorDataTransferObject;
+
+/**
+ * Encapsulates result of {@link VerifyBackupPartitionsTaskV2}.
+ */
+public class IdleVerifyResultV2 extends VisorDataTransferObject {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Counter conflicts. */
+ private Map<PartitionKeyV2, List<PartitionHashRecordV2>> cntrConflicts;
+
+ /** Hash conflicts. */
+ private Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashConflicts;
+
+ /** Moving partitions. */
+ private Map<PartitionKeyV2, List<PartitionHashRecordV2>> movingPartitions;
+
+ /**
+ * @param cntrConflicts Counter conflicts.
+ * @param hashConflicts Hash conflicts.
+ * @param movingPartitions Moving partitions.
+ */
+ public IdleVerifyResultV2(
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> cntrConflicts,
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashConflicts,
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> movingPartitions
+ ) {
+ this.cntrConflicts = cntrConflicts;
+ this.hashConflicts = hashConflicts;
+ this.movingPartitions = movingPartitions;
+ }
+
+ /**
+ * Default constructor for Externalizable.
+ */
+ public IdleVerifyResultV2() {
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void writeExternalData(ObjectOutput out) throws IOException {
+ U.writeMap(out, cntrConflicts);
+ U.writeMap(out, hashConflicts);
+ U.writeMap(out, movingPartitions);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void readExternalData(byte protoVer,
+ ObjectInput in) throws IOException, ClassNotFoundException {
+ cntrConflicts = U.readMap(in);
+ hashConflicts = U.readMap(in);
+ movingPartitions = U.readMap(in);
+ }
+
+ /**
+ * @return Counter conflicts.
+ */
+ public Map<PartitionKeyV2, List<PartitionHashRecordV2>> counterConflicts() {
+ return cntrConflicts;
+ }
+
+ /**
+ * @return Hash conflicts.
+ */
+ public Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashConflicts() {
+ return hashConflicts;
+ }
+
+ /**
+ * @return Moving partitions.
+ */
+ public Map<PartitionKeyV2, List<PartitionHashRecordV2>> movingPartitions() {
+ return movingPartitions;
+ }
+
+ /**
+ * @return <code>true</code> if any conflicts were discovered during idle_verify check.
+ */
+ public boolean hasConflicts() {
+ return !F.isEmpty(hashConflicts()) || !F.isEmpty(counterConflicts());
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(IdleVerifyResultV2.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/PartitionHashRecordV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/PartitionHashRecordV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/PartitionHashRecordV2.java
new file mode 100644
index 0000000..c0f8121
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/PartitionHashRecordV2.java
@@ -0,0 +1,168 @@
+/*
+* 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.processors.cache.verify;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.visor.VisorDataTransferObject;
+
+/**
+ * Record containing partition checksum, primary flag and consistent ID of owner.
+ */
+public class PartitionHashRecordV2 extends VisorDataTransferObject {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Moving partition size. */
+ public static final long MOVING_PARTITION_SIZE = Long.MIN_VALUE;
+
+ /** Partition key. */
+ @GridToStringExclude
+ private PartitionKeyV2 partKey;
+
+ /** Is primary flag. */
+ private boolean isPrimary;
+
+ /** Consistent id. */
+ @GridToStringInclude
+ private Object consistentId;
+
+ /** Partition hash. */
+ @GridToStringExclude
+ private int partHash;
+
+ /** Update counter. */
+ private long updateCntr;
+
+ /** Size. */
+ @GridToStringExclude
+ private long size;
+
+ /**
+ * @param partKey Partition key.
+ * @param isPrimary Is primary.
+ * @param consistentId Consistent id.
+ * @param partHash Partition hash.
+ * @param updateCntr Update counter.
+ * @param size Size.
+ */
+ public PartitionHashRecordV2(PartitionKeyV2 partKey, boolean isPrimary,
+ Object consistentId, int partHash, long updateCntr, long size) {
+ this.partKey = partKey;
+ this.isPrimary = isPrimary;
+ this.consistentId = consistentId;
+ this.partHash = partHash;
+ this.updateCntr = updateCntr;
+ this.size = size;
+ }
+
+ /**
+ * Default constructor for Externalizable.
+ */
+ public PartitionHashRecordV2() {
+ }
+
+ /**
+ * @return Partition key.
+ */
+ public PartitionKeyV2 partitionKey() {
+ return partKey;
+ }
+
+ /**
+ * @return Is primary.
+ */
+ public boolean isPrimary() {
+ return isPrimary;
+ }
+
+ /**
+ * @return Consistent id.
+ */
+ public Object consistentId() {
+ return consistentId;
+ }
+
+ /**
+ * @return Partition hash.
+ */
+ public int partitionHash() {
+ return partHash;
+ }
+
+ /**
+ * @return Update counter.
+ */
+ public long updateCounter() {
+ return updateCntr;
+ }
+
+ /**
+ * @return Size.
+ */
+ public long size() {
+ return size;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void writeExternalData(ObjectOutput out) throws IOException {
+ out.writeObject(partKey);
+ out.writeBoolean(isPrimary);
+ out.writeObject(consistentId);
+ out.writeInt(partHash);
+ out.writeLong(updateCntr);
+ out.writeLong(size);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void readExternalData(byte protoVer, ObjectInput in) throws IOException, ClassNotFoundException {
+ partKey = (PartitionKeyV2)in.readObject();
+ isPrimary = in.readBoolean();
+ consistentId = in.readObject();
+ partHash = in.readInt();
+ updateCntr = in.readLong();
+ size = in.readLong();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return size == MOVING_PARTITION_SIZE ?
+ S.toString(PartitionHashRecordV2.class, this, "state", "MOVING") :
+ S.toString(PartitionHashRecordV2.class, this, "size", size, "partHash", partHash);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ PartitionHashRecordV2 record = (PartitionHashRecordV2)o;
+
+ return consistentId.equals(record.consistentId);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return consistentId.hashCode();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/PartitionKeyV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/PartitionKeyV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/PartitionKeyV2.java
new file mode 100644
index 0000000..1332b0a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/PartitionKeyV2.java
@@ -0,0 +1,127 @@
+/*
+* 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.processors.cache.verify;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.internal.visor.VisorDataTransferObject;
+
+/**
+ * Partition key - pair of cache group ID and partition ID.
+ */
+public class PartitionKeyV2 extends VisorDataTransferObject {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Group id. */
+ private int grpId;
+
+ /** Group name. Optional field, used only for output. */
+ private volatile String grpName;
+
+ /** Partition id. */
+ private int partId;
+
+ /**
+ * @param grpId Group id.
+ * @param partId Partition id.
+ * @param grpName Group name.
+ */
+ public PartitionKeyV2(int grpId, int partId, String grpName) {
+ this.grpId = grpId;
+ this.partId = partId;
+ this.grpName = grpName;
+ }
+
+ /**
+ * Default constructor for Externalizable.
+ */
+ public PartitionKeyV2() {
+ }
+
+ /**
+ * @return Group id.
+ */
+ public int groupId() {
+ return grpId;
+ }
+
+ /**
+ * @return Partition id.
+ */
+ public int partitionId() {
+ return partId;
+ }
+
+ /**
+ * @return Group name.
+ */
+ public String groupName() {
+ return grpName;
+ }
+
+ /**
+ * @param grpName Group name.
+ */
+ public void groupName(String grpName) {
+ this.grpName = grpName;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void writeExternalData(ObjectOutput out) throws IOException {
+ out.writeInt(grpId);
+ U.writeString(out, grpName);
+ out.writeInt(partId);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void readExternalData(byte protoVer, ObjectInput in) throws IOException, ClassNotFoundException {
+ grpId = in.readInt();
+ grpName = U.readString(in);
+ partId = in.readInt();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ PartitionKeyV2 key = (PartitionKeyV2)o;
+
+ return grpId == key.grpId && partId == key.partId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = grpId;
+
+ res = 31 * res + partId;
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(PartitionKeyV2.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTask.java
index b884cb0..c996818 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTask.java
@@ -67,8 +67,11 @@ import org.jetbrains.annotations.Nullable;
* <br>
* Works properly only on idle cluster - there may be false positive conflict reports if data in cluster is being
* concurrently updated.
+ *
+ * @deprecated Legacy version of {@link VerifyBackupPartitionsTaskV2}.
*/
@GridInternal
+@Deprecated
public class VerifyBackupPartitionsTask extends ComputeTaskAdapter<Set<String>,
Map<PartitionKey, List<PartitionHashRecord>>> {
/** */
@@ -154,9 +157,10 @@ public class VerifyBackupPartitionsTask extends ComputeTaskAdapter<Set<String>,
}
/**
- *
+ * Legacy version of {@link VerifyBackupPartitionsTaskV2} internal job, kept for compatibility.
*/
- public static class VerifyBackupPartitionsJob extends ComputeJobAdapter {
+ @Deprecated
+ private static class VerifyBackupPartitionsJob extends ComputeJobAdapter {
/** */
private static final long serialVersionUID = 0L;
@@ -177,7 +181,7 @@ public class VerifyBackupPartitionsTask extends ComputeTaskAdapter<Set<String>,
/**
* @param names Names.
*/
- private VerifyBackupPartitionsJob(Set<String> names) {
+ public VerifyBackupPartitionsJob(Set<String> names) {
cacheNames = names;
}
@@ -244,7 +248,7 @@ public class VerifyBackupPartitionsTask extends ComputeTaskAdapter<Set<String>,
Future<Map<PartitionKey, PartitionHashRecord>> fut = partHashCalcFutures.get(i);
try {
- Map<PartitionKey, PartitionHashRecord> partHash = fut.get(10, TimeUnit.SECONDS);
+ Map<PartitionKey, PartitionHashRecord> partHash = fut.get(100, TimeUnit.MILLISECONDS);
res.putAll(partHash);
@@ -261,7 +265,7 @@ public class VerifyBackupPartitionsTask extends ComputeTaskAdapter<Set<String>,
else
throw new IgniteException(e.getCause());
}
- catch (TimeoutException e) {
+ catch (TimeoutException ignored) {
if (U.currentTimeMillis() - lastProgressLogTs > 3 * 60 * 1000L) {
lastProgressLogTs = U.currentTimeMillis();
@@ -289,7 +293,6 @@ public class VerifyBackupPartitionsTask extends ComputeTaskAdapter<Set<String>,
});
}
-
/**
* @param grpCtx Group context.
* @param part Local partition.
@@ -355,5 +358,4 @@ public class VerifyBackupPartitionsTask extends ComputeTaskAdapter<Set<String>,
return Collections.singletonMap(partKey, partRec);
}
}
-
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTaskV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTaskV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTaskV2.java
new file mode 100644
index 0000000..826393a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTaskV2.java
@@ -0,0 +1,347 @@
+/*
+* 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.processors.cache.verify;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteInterruptedException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobAdapter;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.task.GridInternal;
+import org.apache.ignite.internal.util.lang.GridIterator;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.internal.visor.verify.VisorIdleVerifyTaskArg;
+import org.apache.ignite.lang.IgniteProductVersion;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.resources.LoggerResource;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Task for comparing update counters and checksums between primary and backup partitions of specified caches.
+ * <br>
+ * Argument: Set of cache names, 'null' will trigger verification for all caches.
+ * <br>
+ * Result: {@link IdleVerifyResultV2} with conflict partitions.
+ * <br>
+ * Works properly only on idle cluster - there may be false positive conflict reports if data in cluster is being
+ * concurrently updated.
+ */
+@GridInternal
+public class VerifyBackupPartitionsTaskV2 extends ComputeTaskAdapter<VisorIdleVerifyTaskArg, IdleVerifyResultV2> {
+ /** First version of Ignite that is capable of executing Idle Verify V2. */
+ public static final IgniteProductVersion V2_SINCE_VER = IgniteProductVersion.fromString("2.5.3");
+
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** {@inheritDoc} */
+ @Nullable @Override public Map<? extends ComputeJob, ClusterNode> map(
+ List<ClusterNode> subgrid, VisorIdleVerifyTaskArg arg) throws IgniteException {
+ Map<ComputeJob, ClusterNode> jobs = new HashMap<>();
+
+ for (ClusterNode node : subgrid)
+ jobs.put(new VerifyBackupPartitionsJobV2(arg), node);
+
+ return jobs;
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public IdleVerifyResultV2 reduce(List<ComputeJobResult> results)
+ throws IgniteException {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> clusterHashes = new HashMap<>();
+
+ for (ComputeJobResult res : results) {
+ Map<PartitionKeyV2, PartitionHashRecordV2> nodeHashes = res.getData();
+
+ for (Map.Entry<PartitionKeyV2, PartitionHashRecordV2> e : nodeHashes.entrySet()) {
+ List<PartitionHashRecordV2> records = clusterHashes.computeIfAbsent(e.getKey(), k -> new ArrayList<>());
+
+ records.add(e.getValue());
+ }
+ }
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashConflicts = new HashMap<>();
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> updateCntrConflicts = new HashMap<>();
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> movingParts = new HashMap<>();
+
+ for (Map.Entry<PartitionKeyV2, List<PartitionHashRecordV2>> e : clusterHashes.entrySet()) {
+ Integer partHash = null;
+ Long updateCntr = null;
+
+ for (PartitionHashRecordV2 record : e.getValue()) {
+ if (record.size() == PartitionHashRecordV2.MOVING_PARTITION_SIZE) {
+ List<PartitionHashRecordV2> records = movingParts.computeIfAbsent(
+ e.getKey(), k -> new ArrayList<>());
+
+ records.add(record);
+
+ continue;
+ }
+
+ if (partHash == null) {
+ partHash = record.partitionHash();
+
+ updateCntr = record.updateCounter();
+ }
+ else {
+ if (record.updateCounter() != updateCntr)
+ updateCntrConflicts.putIfAbsent(e.getKey(), e.getValue());
+
+ if (record.partitionHash() != partHash)
+ hashConflicts.putIfAbsent(e.getKey(), e.getValue());
+ }
+ }
+ }
+
+ return new IdleVerifyResultV2(updateCntrConflicts, hashConflicts, movingParts);
+ }
+
+ /**
+ * Job that collects update counters and hashes of local partitions.
+ */
+ private static class VerifyBackupPartitionsJobV2 extends ComputeJobAdapter {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Ignite instance. */
+ @IgniteInstanceResource
+ private IgniteEx ignite;
+
+ /** Injected logger. */
+ @LoggerResource
+ private IgniteLogger log;
+
+ /** Idle verify arguments. */
+ private VisorIdleVerifyTaskArg arg;
+
+ /** Counter of processed partitions. */
+ private final AtomicInteger completionCntr = new AtomicInteger(0);
+
+ /**
+ * @param arg Argument.
+ */
+ public VerifyBackupPartitionsJobV2(VisorIdleVerifyTaskArg arg) {
+ this.arg = arg;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Map<PartitionKeyV2, PartitionHashRecordV2> execute() throws IgniteException {
+ Set<Integer> grpIds = new HashSet<>();
+
+ Set<String> missingCaches = new HashSet<>();
+
+ if (arg.getCaches() != null) {
+ for (String cacheName : arg.getCaches()) {
+ DynamicCacheDescriptor desc = ignite.context().cache().cacheDescriptor(cacheName);
+
+ if (desc == null) {
+ missingCaches.add(cacheName);
+
+ continue;
+ }
+
+ grpIds.add(desc.groupId());
+ }
+
+ if (!missingCaches.isEmpty()) {
+ StringBuilder strBuilder = new StringBuilder("The following caches do not exist: ");
+
+ for (String name : missingCaches)
+ strBuilder.append(name).append(", ");
+
+ strBuilder.delete(strBuilder.length() - 2, strBuilder.length());
+
+ throw new IgniteException(strBuilder.toString());
+ }
+ }
+ else {
+ Collection<CacheGroupContext> groups = ignite.context().cache().cacheGroups();
+
+ for (CacheGroupContext grp : groups) {
+ if (!grp.systemCache() && !grp.isLocal())
+ grpIds.add(grp.groupId());
+ }
+ }
+
+ List<Future<Map<PartitionKeyV2, PartitionHashRecordV2>>> partHashCalcFutures = new ArrayList<>();
+
+ completionCntr.set(0);
+
+ for (Integer grpId : grpIds) {
+ CacheGroupContext grpCtx = ignite.context().cache().cacheGroup(grpId);
+
+ if (grpCtx == null)
+ continue;
+
+ List<GridDhtLocalPartition> parts = grpCtx.topology().localPartitions();
+
+ for (GridDhtLocalPartition part : parts)
+ partHashCalcFutures.add(calculatePartitionHashAsync(grpCtx, part));
+ }
+
+ Map<PartitionKeyV2, PartitionHashRecordV2> res = new HashMap<>();
+
+ long lastProgressLogTs = U.currentTimeMillis();
+
+ for (int i = 0; i < partHashCalcFutures.size(); ) {
+ Future<Map<PartitionKeyV2, PartitionHashRecordV2>> fut = partHashCalcFutures.get(i);
+
+ try {
+ Map<PartitionKeyV2, PartitionHashRecordV2> partHash = fut.get(100, TimeUnit.MILLISECONDS);
+
+ res.putAll(partHash);
+
+ i++;
+ }
+ catch (InterruptedException | ExecutionException e) {
+ for (int j = i + 1; j < partHashCalcFutures.size(); j++)
+ partHashCalcFutures.get(j).cancel(false);
+
+ if (e instanceof InterruptedException)
+ throw new IgniteInterruptedException((InterruptedException)e);
+ else if (e.getCause() instanceof IgniteException)
+ throw (IgniteException)e.getCause();
+ else
+ throw new IgniteException(e.getCause());
+ }
+ catch (TimeoutException ignored) {
+ if (U.currentTimeMillis() - lastProgressLogTs > 3 * 60 * 1000L) {
+ lastProgressLogTs = U.currentTimeMillis();
+
+ log.warning("idle_verify is still running, processed " + completionCntr.get() + " of " +
+ partHashCalcFutures.size() + " local partitions");
+ }
+ }
+ }
+
+ return res;
+ }
+
+ /**
+ * @param grpCtx Group context.
+ * @param part Local partition.
+ */
+ private Future<Map<PartitionKeyV2, PartitionHashRecordV2>> calculatePartitionHashAsync(
+ final CacheGroupContext grpCtx,
+ final GridDhtLocalPartition part
+ ) {
+ return ForkJoinPool.commonPool().submit(new Callable<Map<PartitionKeyV2, PartitionHashRecordV2>>() {
+ @Override public Map<PartitionKeyV2, PartitionHashRecordV2> call() throws Exception {
+ return calculatePartitionHash(grpCtx, part);
+ }
+ });
+ }
+
+
+ /**
+ * @param grpCtx Group context.
+ * @param part Local partition.
+ */
+ private Map<PartitionKeyV2, PartitionHashRecordV2> calculatePartitionHash(
+ CacheGroupContext grpCtx,
+ GridDhtLocalPartition part
+ ) {
+ if (!part.reserve())
+ return Collections.emptyMap();
+
+ int partHash = 0;
+ long partSize;
+ long updateCntrBefore = part.updateCounter();
+
+ PartitionKeyV2 partKey = new PartitionKeyV2(grpCtx.groupId(), part.id(), grpCtx.cacheOrGroupName());
+
+ Object consId = ignite.context().discovery().localNode().consistentId();
+
+ boolean isPrimary = part.primary(grpCtx.topology().readyTopologyVersion());
+
+ try {
+ if (part.state() == GridDhtPartitionState.MOVING) {
+ PartitionHashRecordV2 movingHashRecord = new PartitionHashRecordV2(partKey, isPrimary, consId,
+ partHash, updateCntrBefore, PartitionHashRecordV2.MOVING_PARTITION_SIZE);
+
+ return Collections.singletonMap(partKey, movingHashRecord);
+ }
+ else if (part.state() != GridDhtPartitionState.OWNING)
+ return Collections.emptyMap();
+
+ partSize = part.dataStore().fullSize();
+
+ GridIterator<CacheDataRow> it = grpCtx.offheap().partitionIterator(part.id());
+
+ while (it.hasNextX()) {
+ CacheDataRow row = it.nextX();
+
+ partHash += row.key().hashCode();
+
+ partHash += Arrays.hashCode(row.value().valueBytes(grpCtx.cacheObjectContext()));
+ }
+
+ long updateCntrAfter = part.updateCounter();
+
+ if (updateCntrBefore != updateCntrAfter) {
+ throw new IgniteException("Cluster is not idle: update counter of partition [grpId=" +
+ grpCtx.groupId() + ", partId=" + part.id() + "] changed during hash calculation [before=" +
+ updateCntrBefore + ", after=" + updateCntrAfter + "]");
+ }
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Can't calculate partition hash [grpId=" + grpCtx.groupId() +
+ ", partId=" + part.id() + "]", e);
+
+ return Collections.emptyMap();
+ }
+ finally {
+ part.release();
+ }
+
+ PartitionHashRecordV2 partRec = new PartitionHashRecordV2(
+ partKey, isPrimary, consId, partHash, updateCntrBefore, partSize);
+
+ completionCntr.incrementAndGet();
+
+ return Collections.singletonMap(partKey, partRec);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/main/java/org/apache/ignite/internal/visor/verify/VisorIdleVerifyTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/verify/VisorIdleVerifyTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/verify/VisorIdleVerifyTask.java
index 05f2621..e67c681 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/verify/VisorIdleVerifyTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/verify/VisorIdleVerifyTask.java
@@ -37,6 +37,7 @@ import org.apache.ignite.resources.JobContextResource;
* Task to verify checksums of backup partitions.
*/
@GridInternal
+@Deprecated
public class VisorIdleVerifyTask extends VisorOneNodeTask<VisorIdleVerifyTaskArg, VisorIdleVerifyTaskResult> {
/** */
private static final long serialVersionUID = 0L;
@@ -49,6 +50,7 @@ public class VisorIdleVerifyTask extends VisorOneNodeTask<VisorIdleVerifyTaskArg
/**
*
*/
+ @Deprecated
private static class VisorIdleVerifyJob extends VisorJob<VisorIdleVerifyTaskArg, VisorIdleVerifyTaskResult> {
/** */
private static final long serialVersionUID = 0L;
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/main/java/org/apache/ignite/internal/visor/verify/VisorIdleVerifyTaskV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/verify/VisorIdleVerifyTaskV2.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/verify/VisorIdleVerifyTaskV2.java
new file mode 100644
index 0000000..b9250ef
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/verify/VisorIdleVerifyTaskV2.java
@@ -0,0 +1,94 @@
+/*
+ * 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.visor.verify;
+
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.compute.ComputeJobContext;
+import org.apache.ignite.compute.ComputeTaskFuture;
+import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
+import org.apache.ignite.internal.processors.cache.verify.VerifyBackupPartitionsTaskV2;
+import org.apache.ignite.internal.processors.task.GridInternal;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.visor.VisorJob;
+import org.apache.ignite.internal.visor.VisorOneNodeTask;
+import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.resources.JobContextResource;
+
+/**
+ * Task to verify checksums of backup partitions.
+ */
+@GridInternal
+public class VisorIdleVerifyTaskV2 extends VisorOneNodeTask<VisorIdleVerifyTaskArg, IdleVerifyResultV2> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** {@inheritDoc} */
+ @Override protected VisorJob<VisorIdleVerifyTaskArg, IdleVerifyResultV2> job(VisorIdleVerifyTaskArg arg) {
+ return new VisorIdleVerifyJobV2(arg, debug);
+ }
+
+ /**
+ *
+ */
+ private static class VisorIdleVerifyJobV2 extends VisorJob<VisorIdleVerifyTaskArg, IdleVerifyResultV2> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private ComputeTaskFuture<IdleVerifyResultV2> fut;
+
+ /** Auto-inject job context. */
+ @JobContextResource
+ protected transient ComputeJobContext jobCtx;
+
+ /**
+ * @param arg Argument.
+ * @param debug Debug.
+ */
+ private VisorIdleVerifyJobV2(VisorIdleVerifyTaskArg arg, boolean debug) {
+ super(arg, debug);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IdleVerifyResultV2 run(VisorIdleVerifyTaskArg arg) throws IgniteException {
+ if (fut == null) {
+ fut = ignite.compute().executeAsync(VerifyBackupPartitionsTaskV2.class, arg);
+
+ if (!fut.isDone()) {
+ jobCtx.holdcc();
+
+ fut.listen(new IgniteInClosure<IgniteFuture<IdleVerifyResultV2>>() {
+ @Override public void apply(IgniteFuture<IdleVerifyResultV2> f) {
+ jobCtx.callcc();
+ }
+ });
+
+ return null;
+ }
+ }
+
+ return fut.get();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(VisorIdleVerifyJobV2.class, this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/main/resources/META-INF/classnames.properties
----------------------------------------------------------------------
diff --git a/modules/core/src/main/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties
index a6f4e46..c72f570 100644
--- a/modules/core/src/main/resources/META-INF/classnames.properties
+++ b/modules/core/src/main/resources/META-INF/classnames.properties
@@ -1136,6 +1136,7 @@ org.apache.ignite.internal.processors.cache.verify.CollectConflictPartitionKeysT
org.apache.ignite.internal.processors.cache.verify.CollectConflictPartitionKeysTask$CollectPartitionEntryHashesJob
org.apache.ignite.internal.processors.cache.verify.ContentionClosure
org.apache.ignite.internal.processors.cache.verify.ContentionInfo
+org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2
org.apache.ignite.internal.processors.cache.verify.PartitionEntryHashRecord
org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord
org.apache.ignite.internal.processors.cache.verify.PartitionKey
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/main/resources/ignite.properties
----------------------------------------------------------------------
diff --git a/modules/core/src/main/resources/ignite.properties b/modules/core/src/main/resources/ignite.properties
index e70660b..785f962 100644
--- a/modules/core/src/main/resources/ignite.properties
+++ b/modules/core/src/main/resources/ignite.properties
@@ -15,7 +15,7 @@
# limitations under the License.
#
-ignite.version=2.5.0-SNAPSHOT
+ignite.version=2.6.0-SNAPSHOT
ignite.build=0
ignite.revision=DEV
ignite.rel.date=01011970
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
index b992f5c..6803ab3 100755
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
@@ -804,10 +804,8 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
Map<PartitionKey, List<PartitionHashRecord>> conflicts = ig.compute().execute(
new VerifyBackupPartitionsTask(), cacheNames);
- if (!conflicts.isEmpty()) {
- throw new IgniteCheckedException("Partition checksums are different for backups " +
- "of the following partitions: " + conflicts.keySet());
- }
+ if (!conflicts.isEmpty())
+ throw new IgniteCheckedException("Conflict partitions: " + conflicts.keySet());
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/9131e4dd/modules/core/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java b/modules/core/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
index 0ced430..11389bf 100644
--- a/modules/core/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
@@ -48,7 +48,14 @@ import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.commandline.CommandHandler;
import org.apache.ignite.internal.commandline.cache.CacheCommand;
+import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
+import org.apache.ignite.internal.processors.cache.CacheObjectImpl;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheFuture;
+import org.apache.ignite.internal.processors.cache.GridCacheOperation;
+import org.apache.ignite.internal.processors.cache.KeyCacheObjectImpl;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.X;
@@ -603,6 +610,77 @@ public class GridCommandHandlerTest extends GridCommonAbstractTest {
}
/**
+ * Tests that both update counter and hash conflicts are detected.
+ *
+ * @throws Exception If failed.
+ */
+ public void testCacheIdleVerifyTwoConflictTypes() throws Exception {
+ IgniteEx ignite = (IgniteEx)startGrids(2);
+
+ ignite.cluster().active(true);
+
+ int parts = 32;
+
+ IgniteCache<Object, Object> cache = ignite.createCache(new CacheConfiguration<>()
+ .setAffinity(new RendezvousAffinityFunction(false, parts))
+ .setBackups(1)
+ .setName(DEFAULT_CACHE_NAME));
+
+ for (int i = 0; i < 100; i++)
+ cache.put(i, i);
+
+ injectTestSystemOut();
+
+ assertEquals(EXIT_CODE_OK, execute("--cache", "idle_verify"));
+
+ assertTrue(testOut.toString().contains("no conflicts have been found"));
+
+ GridCacheContext<Object, Object> cacheCtx = ignite.cachex(DEFAULT_CACHE_NAME).context();
+
+ corruptDataEntry(cacheCtx, 1, true, false);
+
+ corruptDataEntry(cacheCtx, 1 + parts / 2, false, true);
+
+ assertEquals(EXIT_CODE_OK, execute("--cache", "idle_verify"));
+
+ assertTrue(testOut.toString().contains("found 2 conflict partitions"));
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testCacheIdleVerifyMovingParts() throws Exception {
+ IgniteEx ignite = (IgniteEx)startGrids(2);
+
+ ignite.cluster().active(true);
+
+ int parts = 32;
+
+ IgniteCache<Object, Object> cache = ignite.createCache(new CacheConfiguration<>()
+ .setAffinity(new RendezvousAffinityFunction(false, parts))
+ .setBackups(1)
+ .setName(DEFAULT_CACHE_NAME)
+ .setRebalanceDelay(10_000));
+
+ for (int i = 0; i < 100; i++)
+ cache.put(i, i);
+
+ injectTestSystemOut();
+
+ assertEquals(EXIT_CODE_OK, execute("--cache", "idle_verify"));
+
+ assertTrue(testOut.toString().contains("no conflicts have been found"));
+
+ startGrid(2);
+
+ resetBaselineTopology();
+
+ assertEquals(EXIT_CODE_OK, execute("--cache", "idle_verify"));
+
+ assertTrue(testOut.toString().contains("MOVING partitions"));
+ }
+
+ /**
*
*/
public void testCacheContention() throws Exception {
@@ -925,4 +1003,60 @@ public class GridCommandHandlerTest extends GridCommonAbstractTest {
assertTrue("Expecting no active futures: node=" + ig.localNode().id(), futs.isEmpty());
}
}
+
+ /**
+ * Corrupts data entry.
+ *
+ * @param ctx Context.
+ * @param key Key.
+ * @param breakCntr Break counter.
+ * @param breakData Break data.
+ */
+ private void corruptDataEntry(
+ GridCacheContext<Object, Object> ctx,
+ int key,
+ boolean breakCntr,
+ boolean breakData
+ ) {
+ int partId = ctx.affinity().partition(key);
+
+ try {
+ long updateCntr = ctx.topology().localPartition(partId).updateCounter();
+
+ Object valToPut = ctx.cache().keepBinary().get(key);
+
+ if (breakCntr)
+ updateCntr++;
+
+ if (breakData)
+ valToPut = valToPut.toString() + " broken";
+
+ // Create data entry
+ DataEntry dataEntry = new DataEntry(
+ ctx.cacheId(),
+ new KeyCacheObjectImpl(key, null, partId),
+ new CacheObjectImpl(valToPut, null),
+ GridCacheOperation.UPDATE,
+ new GridCacheVersion(),
+ new GridCacheVersion(),
+ 0L,
+ partId,
+ updateCntr
+ );
+
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)ctx.shared().database();
+
+ db.checkpointReadLock();
+
+ try {
+ U.invoke(GridCacheDatabaseSharedManager.class, db, "applyUpdate", ctx, dataEntry);
+ }
+ finally {
+ db.checkpointReadUnlock();
+ }
+ }
+ catch (IgniteCheckedException e) {
+ e.printStackTrace();
+ }
+ }
}