You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2023/01/13 10:45:50 UTC
[ignite-3] branch main updated: IGNITE-18542 Remove some server-side MetaStorage classes (#1518)
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 174138bb72 IGNITE-18542 Remove some server-side MetaStorage classes (#1518)
174138bb72 is described below
commit 174138bb7293003ef94a5b25ce3b6192c78fe7f4
Author: Alexander Polovtcev <al...@gmail.com>
AuthorDate: Fri Jan 13 13:45:44 2023 +0300
IGNITE-18542 Remove some server-side MetaStorage classes (#1518)
---
...butionZoneManagerLogicalTopologyEventsTest.java | 2 +-
.../DistributionZoneManagerWatchListenerTest.java | 4 +-
.../apache/ignite/internal/metastorage/Entry.java | 4 +-
.../ignite/internal/metastorage/EntryEvent.java | 11 +-
.../internal/metastorage/dsl/Conditions.java | 21 +-
.../ignite/internal/metastorage/dsl/Operation.java | 168 +++----
.../internal/metastorage/dsl/Operations.java | 8 +-
.../internal/metastorage/dsl/SimpleCondition.java | 400 ++++++-----------
.../impl/ItMetaStorageServicePersistenceTest.java | 23 +-
.../metastorage/impl/ItMetaStorageServiceTest.java | 159 +++----
.../server/raft/ItMetaStorageRaftGroupTest.java | 30 +-
.../metastorage/command/SingleEntryResponse.java | 11 +-
.../internal/metastorage/impl/EntryImpl.java | 70 ++-
.../metastorage/impl/MetaStorageService.java | 73 +---
.../metastorage/impl/MetaStorageServiceImpl.java | 117 +++--
.../server/AbstractCompoundCondition.java | 1 +
.../server/AbstractSimpleCondition.java | 1 +
.../internal/metastorage/server/Condition.java | 6 +-
.../ignite/internal/metastorage/server/Entry.java | 192 --------
.../internal/metastorage/server/EntryEvent.java | 58 ---
.../metastorage/server/ExistenceCondition.java | 1 +
.../ignite/internal/metastorage/server/If.java | 2 +
.../metastorage/server/KeyValueStorage.java | 23 +-
.../internal/metastorage/server/Operation.java | 104 -----
.../metastorage/server/RevisionCondition.java | 1 +
.../internal/metastorage/server/Statement.java | 2 +
.../metastorage/server/StatementResult.java | 65 ---
.../metastorage/server/TombstoneCondition.java | 1 +
.../ignite/internal/metastorage/server/Update.java | 60 ---
.../metastorage/server/ValueCondition.java | 1 +
.../internal/metastorage/server/WatchEvent.java | 72 ---
.../server/persistence/RangeCursor.java | 2 +-
.../server/persistence/RocksDbKeyValueStorage.java | 23 +-
.../server/persistence/WatchCursor.java | 11 +-
.../server/raft/MetaStorageListener.java | 14 +-
.../metastorage/watch/WatchAggregator.java | 4 +-
.../impl/MetaStorageRangeCursorTest.java | 10 +-
.../server/AbstractKeyValueStorageTest.java | 481 ++++++++++-----------
.../metastorage/server/AndConditionTest.java | 11 +-
.../metastorage/server/ExistenceConditionTest.java | 8 +-
.../metastorage/server/OrConditionTest.java | 11 +-
.../metastorage/server/RevisionConditionTest.java | 17 +-
.../metastorage/server/TombstoneConditionTest.java | 8 +-
.../metastorage/server/ValueConditionTest.java | 29 +-
.../metastorage/watch/WatchAggregatorTest.java | 4 +-
.../server/SimpleInMemoryKeyValueStorage.java | 22 +-
.../storage/DistributedConfigurationStorage.java | 19 +-
.../DistributedConfigurationCatchUpTest.java | 4 +-
.../DistributedConfigurationStorageTest.java | 115 +----
.../ignite/internal/schema/SchemaManager.java | 28 +-
.../internal/table/distributed/TableManager.java | 2 +-
.../ignite/internal/utils/RebalanceUtil.java | 11 +-
52 files changed, 808 insertions(+), 1717 deletions(-)
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java
index f4cec188fd..b8acc1ff63 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java
@@ -265,7 +265,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
return metaStorageService.run(getCommand).thenApply(bi -> {
SingleEntryResponse resp = (SingleEntryResponse) bi;
- return new EntryImpl(new ByteArray(resp.key()), resp.value(), resp.revision(), resp.updateCounter());
+ return new EntryImpl(resp.key(), resp.value(), resp.revision(), resp.updateCounter());
});
}).when(metaStorageManager).get(any());
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java
index e4406e410b..5828e7fbc3 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java
@@ -57,6 +57,7 @@ import org.apache.ignite.internal.distributionzones.configuration.DistributionZo
import org.apache.ignite.internal.distributionzones.configuration.DistributionZoneConfiguration;
import org.apache.ignite.internal.distributionzones.configuration.DistributionZoneView;
import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
+import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.EntryEvent;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.WatchEvent;
@@ -68,7 +69,6 @@ import org.apache.ignite.internal.metastorage.dsl.If;
import org.apache.ignite.internal.metastorage.dsl.StatementResult;
import org.apache.ignite.internal.metastorage.impl.EntryImpl;
import org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
-import org.apache.ignite.internal.metastorage.server.Entry;
import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageListener;
import org.apache.ignite.internal.raft.Command;
@@ -448,7 +448,7 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
private void watchListenerOnUpdate(Set<String> nodes, long rev) {
byte[] newLogicalTopology = toBytes(nodes);
- org.apache.ignite.internal.metastorage.Entry newEntry = new EntryImpl(zonesLogicalTopologyKey(), newLogicalTopology, rev, 1);
+ Entry newEntry = new EntryImpl(zonesLogicalTopologyKey().bytes(), newLogicalTopology, rev, 1);
EntryEvent entryEvent = new EntryEvent(null, newEntry);
diff --git a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/Entry.java b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/Entry.java
index 32b0a43ea2..8945b17f28 100644
--- a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/Entry.java
+++ b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/Entry.java
@@ -17,8 +17,6 @@
package org.apache.ignite.internal.metastorage;
-import org.apache.ignite.lang.ByteArray;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -37,7 +35,7 @@ public interface Entry {
*
* @return The key.
*/
- @NotNull ByteArray key();
+ byte[] key();
/**
* Returns a value. Could be {@code null} for empty entry.
diff --git a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/EntryEvent.java b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/EntryEvent.java
index 29ffbe09df..4d2a9c2bca 100644
--- a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/EntryEvent.java
+++ b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/EntryEvent.java
@@ -18,18 +18,16 @@
package org.apache.ignite.internal.metastorage;
import java.io.Serializable;
-import org.jetbrains.annotations.NotNull;
+import org.apache.ignite.internal.tostring.S;
/**
* Represent an update event for particular key and entry.
*/
public final class EntryEvent implements Serializable {
/** Old (previous) entry. */
- @NotNull
private final Entry oldEntry;
/** New (updated) entry. */
- @NotNull
private final Entry newEntry;
/**
@@ -48,7 +46,6 @@ public final class EntryEvent implements Serializable {
*
* @return Old entry.
*/
- @NotNull
public Entry oldEntry() {
return oldEntry;
}
@@ -58,7 +55,6 @@ public final class EntryEvent implements Serializable {
*
* @return New entry.
*/
- @NotNull
public Entry newEntry() {
return newEntry;
}
@@ -92,4 +88,9 @@ public final class EntryEvent implements Serializable {
return res;
}
+
+ @Override
+ public String toString() {
+ return S.toString(this);
+ }
}
diff --git a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Conditions.java b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Conditions.java
index 2424cf322a..052d54dee4 100644
--- a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Conditions.java
+++ b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Conditions.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.metastorage.dsl;
import org.apache.ignite.lang.ByteArray;
-import org.jetbrains.annotations.NotNull;
/**
* This class contains fabric methods which produce conditions needed for a conditional multi update functionality provided by the meta
@@ -34,8 +33,8 @@ public final class Conditions {
* @return Condition on entry revision.
* @see SimpleCondition.RevisionCondition
*/
- public static SimpleCondition.RevisionCondition revision(@NotNull ByteArray key) {
- return new SimpleCondition.RevisionCondition(key.bytes());
+ public static SimpleCondition.RevisionConditionBuilder revision(ByteArray key) {
+ return SimpleCondition.revision(key.bytes());
}
/**
@@ -45,8 +44,8 @@ public final class Conditions {
* @return Condition on entry value.
* @see SimpleCondition.ValueCondition
*/
- public static SimpleCondition.ValueCondition value(@NotNull ByteArray key) {
- return new SimpleCondition.ValueCondition(key.bytes());
+ public static SimpleCondition.ValueConditionBuilder value(ByteArray key) {
+ return SimpleCondition.value(key.bytes());
}
/**
@@ -55,8 +54,8 @@ public final class Conditions {
* @param key Identifies an entry, which condition will be applied to. Can't be {@code null}.
* @return Condition on entry existence.
*/
- public static SimpleCondition exists(@NotNull ByteArray key) {
- return new SimpleCondition.ExistenceCondition(key.bytes()).exists();
+ public static SimpleCondition exists(ByteArray key) {
+ return SimpleCondition.exists(key.bytes());
}
/**
@@ -65,8 +64,8 @@ public final class Conditions {
* @param key Identifies an entry, which condition will be applied to. Can't be {@code null}.
* @return Condition on entry not existence.
*/
- public static SimpleCondition notExists(@NotNull ByteArray key) {
- return new SimpleCondition.ExistenceCondition(key.bytes()).notExists();
+ public static SimpleCondition notExists(ByteArray key) {
+ return SimpleCondition.notExists(key.bytes());
}
/**
@@ -75,8 +74,8 @@ public final class Conditions {
* @param key Identifies an entry, which condition will be applied to. Can't be {@code null}.
* @return Condition on entry's value is tombstone.
*/
- public static SimpleCondition tombstone(@NotNull ByteArray key) {
- return new SimpleCondition.TombstoneCondition(key.bytes()).tombstone();
+ public static SimpleCondition tombstone(ByteArray key) {
+ return SimpleCondition.tombstone(key.bytes());
}
/**
diff --git a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Operation.java b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Operation.java
index 6b229633a8..06c54cd682 100644
--- a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Operation.java
+++ b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Operation.java
@@ -17,158 +17,96 @@
package org.apache.ignite.internal.metastorage.dsl;
-import org.jetbrains.annotations.NotNull;
+import org.apache.ignite.internal.util.IgniteUtils;
import org.jetbrains.annotations.Nullable;
/**
* Defines operation for meta storage conditional update (invoke).
*/
public final class Operation {
- /** Actual operation implementation. */
- private final InnerOp upd;
+ /**
+ * Key identifies an entry which operation will be applied to. Key is {@code null} for {@link OperationType#NO_OP} operation.
+ */
+ private final byte @Nullable [] key;
/**
- * Constructs an operation which wraps the actual operation implementation.
- *
- * @param upd The actual operation implementation.
+ * Value which will be associated with the {@link #key}. Value is not {@code null} only for {@link OperationType#PUT} operation.
*/
- Operation(InnerOp upd) {
- this.upd = upd;
- }
+ private final byte @Nullable [] val;
/**
- * Returns actual operation implementation.
+ * Operation type.
+ *
+ * @see OperationType
*/
- public InnerOp inner() {
- return upd;
- }
+ private final OperationType type;
/**
- * Returns operation type.
+ * Constructs operation which will be applied to an entry identified by the given key.
*
- * @return Operation type.
+ * @param type Operation type. Can't be {@code null}.
+ * @param key Key identifies an entry which operation will be applied to.
+ * @param val Value will be associated with an entry identified by the {@code key}.
*/
- public OperationType type() {
- return upd.type();
+ public Operation(OperationType type, byte @Nullable [] key, byte @Nullable [] val) {
+ assert (type == OperationType.NO_OP && key == null && val == null)
+ || (type == OperationType.PUT && key != null && val != null)
+ || (type == OperationType.REMOVE && key != null && val == null)
+ : "Invalid operation parameters: [type=" + type
+ + ", key=" + (key == null ? "null" : IgniteUtils.toHexString(key, 256))
+ + ", val=" + (val == null ? "null" : IgniteUtils.toHexString(val, 256)) + ']';
+
+ this.key = key;
+ this.val = val;
+ this.type = type;
}
/**
- * Represents operation of type <i>remove</i>.
+ * Returns a key which identifies an entry which operation will be applied to.
+ *
+ * @return A key which identifies an entry which operation will be applied to.
*/
- public static final class RemoveOp extends AbstractOp {
- /**
- * Default no-op constructor.
- *
- * @param key Identifies an entry which operation will be applied to.
- */
- RemoveOp(byte[] key) {
- super(key, OperationType.REMOVE);
- }
+ public byte @Nullable [] key() {
+ return key;
}
/**
- * Represents operation of type <i>put</i>.
+ * Returns a value which will be associated with an entry identified by the {@code key}.
+ *
+ * @return A value which will be associated with an entry identified by the {@code key}.
*/
- public static final class PutOp extends AbstractOp {
- /** Value. */
- private final byte[] val;
-
- /**
- * Constructs operation of type <i>put</i>.
- *
- * @param key Identifies an entry which operation will be applied to.
- * @param val The value to which the entry should be updated.
- */
- PutOp(byte[] key, byte[] val) {
- super(key, OperationType.PUT);
- this.val = val;
- }
-
- /**
- * Returns value.
- *
- * @return Value.
- */
- public byte[] value() {
- return val;
- }
+ public byte @Nullable [] value() {
+ return val;
}
/**
- * Represents operation of type <i>no-op</i>.
+ * Returns an operation type.
+ *
+ * @return An operation type.
*/
- public static final class NoOp extends AbstractOp {
- /**
- * Default no-op constructor.
- */
- NoOp() {
- super(null, OperationType.NO_OP);
- }
+ public OperationType type() {
+ return type;
}
/**
- * Defines operation interface.
+ * Creates an operation of type <i>remove</i>.
*/
- public interface InnerOp {
- /**
- * Returns key.
- *
- * @return Key.
- */
- @Nullable byte[] key();
-
- /**
- * Returns operation type.
- *
- * @return Operation type.
- */
- @NotNull OperationType type();
+ public static Operation remove(byte[] key) {
+ return new Operation(OperationType.REMOVE, key, null);
}
/**
- * Extension of {@link InnerOp}.
+ * Creates an operation of type <i>put</i>.
*/
- private static class AbstractOp implements InnerOp {
- /** Key. */
- @Nullable
- private final byte[] key;
-
- /** Operation type. */
- @NotNull
- private final OperationType type;
-
- /**
- * Ctor.
- *
- * @param key Key.
- * @param type Operation type.
- */
- private AbstractOp(@Nullable byte[] key, OperationType type) {
- this.key = key;
- this.type = type;
- }
-
- /**
- * Returns key.
- *
- * @return Key.
- */
- @Nullable
- @Override
- public byte[] key() {
- return key;
- }
+ public static Operation put(byte[] key, byte[] val) {
+ return new Operation(OperationType.PUT, key, val);
+ }
- /**
- * Returns operation type.
- *
- * @return Operation type.
- */
- @NotNull
- @Override
- public OperationType type() {
- return type;
- }
+ /**
+ * Creates an operation of type <i>no-op</i>.
+ */
+ public static Operation noOp() {
+ return new Operation(OperationType.NO_OP, null, null);
}
}
diff --git a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Operations.java b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Operations.java
index b70a4cc56e..e4f6d26b5a 100644
--- a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Operations.java
+++ b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/Operations.java
@@ -30,7 +30,7 @@ import org.apache.ignite.lang.ByteArray;
*/
public final class Operations {
/** No-op operation singleton. */
- private static final Operation.NoOp NO_OP = new Operation.NoOp();
+ private static final Operation NO_OP = Operation.noOp();
/** Operations. */
private final Operation[] operations;
@@ -90,7 +90,7 @@ public final class Operations {
* @return Operation of type <i>remove</i>.
*/
public static Operation remove(ByteArray key) {
- return new Operation(new Operation.RemoveOp(key.bytes()));
+ return Operation.remove(key.bytes());
}
/**
@@ -101,7 +101,7 @@ public final class Operations {
* @return Operation of type <i>put</i>.
*/
public static Operation put(ByteArray key, byte[] value) {
- return new Operation(new Operation.PutOp(key.bytes(), value));
+ return Operation.put(key.bytes(), value);
}
/**
@@ -110,6 +110,6 @@ public final class Operations {
* @return Operation of type <i>noop</i>.
*/
public static Operation noop() {
- return new Operation(NO_OP);
+ return NO_OP;
}
}
diff --git a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/SimpleCondition.java b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/SimpleCondition.java
index dfe855f79c..a5296a9fe6 100644
--- a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/SimpleCondition.java
+++ b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/dsl/SimpleCondition.java
@@ -20,46 +20,98 @@ package org.apache.ignite.internal.metastorage.dsl;
/**
* Represents a condition for a meta storage conditional update.
*/
-public final class SimpleCondition implements Condition {
- /** Actual condition implementation. */
- private final InnerCondition cond;
+public class SimpleCondition implements Condition {
+ /** Entry key. */
+ private final byte[] key;
/**
- * Constructs a condition, which wraps the actual condition implementation.
- *
- * @param cond The actual condition implementation.
+ * Condition type.
*/
- SimpleCondition(InnerCondition cond) {
- this.cond = cond;
+ private final ConditionType type;
+
+ private SimpleCondition(byte[] key, ConditionType type) {
+ this.key = key;
+ this.type = type;
}
- public InnerCondition inner() {
- return cond;
+ /**
+ * Returns the key, which identifies an entry, which condition will be applied to.
+ *
+ * @return Key, which identifies an entry, which condition will be applied to.
+ */
+ public byte[] key() {
+ return key;
}
+ /**
+ * Returns the condition type.
+ *
+ * @return Condition type.
+ */
public ConditionType type() {
- return cond.type();
+ return type;
}
/**
- * Represents a condition on an entry revision. Only one type of condition could be applied to the one instance of a condition.
- * Subsequent invocations of any method, which produces a condition will throw {@link IllegalStateException}.
+ * Creates a builder for a revision condition.
+ *
+ * @param key Key, which identifies an entry, which condition will be applied to.
+ * @return Builder for a revision condition.
+ * @see RevisionCondition
*/
- public static final class RevisionCondition extends AbstractCondition {
- /** The revision as the condition argument. */
- private long rev;
+ public static RevisionConditionBuilder revision(byte[] key) {
+ return new RevisionConditionBuilder(key);
+ }
- /**
- * Constructs a condition by a revision for an entry identified by the given key.
- *
- * @param key Identifies an entry, which condition will be applied to.
- */
- RevisionCondition(byte[] key) {
- super(key);
- }
+ /**
+ * Creates a builder for a value condition.
+ *
+ * @param key Key, which identifies an entry, which condition will be applied to.
+ * @return Builder for a value condition.
+ * @see ValueCondition
+ */
+ public static ValueConditionBuilder value(byte[] key) {
+ return new ValueConditionBuilder(key);
+ }
- public long revision() {
- return rev;
+ /**
+ * Produces the condition of type {@link ConditionType#TOMBSTONE}. This condition tests that an entry's value, identified by the
+ * given key, is tombstone.
+ *
+ * @return The condition of type {@link ConditionType#TOMBSTONE}.
+ */
+ public static SimpleCondition tombstone(byte[] key) {
+ return new SimpleCondition(key, ConditionType.TOMBSTONE);
+ }
+
+ /**
+ * Produces the condition of type {@link ConditionType#KEY_EXISTS}. This condition tests the existence of an entry identified by the
+ * given key.
+ *
+ * @return The condition of type {@link ConditionType#KEY_EXISTS}.
+ */
+ public static SimpleCondition exists(byte[] key) {
+ return new SimpleCondition(key, ConditionType.KEY_EXISTS);
+ }
+
+ /**
+ * Produces the condition of type {@link ConditionType#KEY_NOT_EXISTS}. This condition tests the non-existence of an entry
+ * identified by the given key.
+ *
+ * @return The condition of type {@link ConditionType#KEY_NOT_EXISTS}.
+ */
+ public static SimpleCondition notExists(byte[] key) {
+ return new SimpleCondition(key, ConditionType.KEY_NOT_EXISTS);
+ }
+
+ /**
+ * Builder for {@link RevisionCondition}.
+ */
+ public static final class RevisionConditionBuilder {
+ private final byte[] key;
+
+ RevisionConditionBuilder(byte[] key) {
+ this.key = key;
}
/**
@@ -68,18 +120,9 @@ public final class SimpleCondition implements Condition {
*
* @param rev The revision.
* @return The condition of type {@link ConditionType#REV_EQUAL}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition eq(long rev) {
- assert rev > 0 : "Revision must be positive.";
-
- validate(type());
-
- type(ConditionType.REV_EQUAL);
-
- this.rev = rev;
-
- return new SimpleCondition(this);
+ return new RevisionCondition(key, ConditionType.REV_EQUAL, rev);
}
/**
@@ -88,18 +131,9 @@ public final class SimpleCondition implements Condition {
*
* @param rev The revision.
* @return The condition of type {@link ConditionType#REV_NOT_EQUAL}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition ne(long rev) {
- assert rev > 0 : "Revision must be positive.";
-
- validate(type());
-
- type(ConditionType.REV_NOT_EQUAL);
-
- this.rev = rev;
-
- return new SimpleCondition(this);
+ return new RevisionCondition(key, ConditionType.REV_NOT_EQUAL, rev);
}
/**
@@ -108,18 +142,9 @@ public final class SimpleCondition implements Condition {
*
* @param rev The revision.
* @return The condition of type {@link ConditionType#REV_GREATER}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition gt(long rev) {
- assert rev > 0 : "Revision must be positive.";
-
- validate(type());
-
- type(ConditionType.REV_GREATER);
-
- this.rev = rev;
-
- return new SimpleCondition(this);
+ return new RevisionCondition(key, ConditionType.REV_GREATER, rev);
}
/**
@@ -128,18 +153,9 @@ public final class SimpleCondition implements Condition {
*
* @param rev The revision.
* @return The condition of type {@link ConditionType#REV_GREATER_OR_EQUAL}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition ge(long rev) {
- assert rev > 0 : "Revision must be positive.";
-
- validate(type());
-
- type(ConditionType.REV_GREATER_OR_EQUAL);
-
- this.rev = rev;
-
- return new SimpleCondition(this);
+ return new RevisionCondition(key, ConditionType.REV_GREATER_OR_EQUAL, rev);
}
/**
@@ -148,18 +164,9 @@ public final class SimpleCondition implements Condition {
*
* @param rev The revision.
* @return The condition of type {@link ConditionType#REV_LESS}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition lt(long rev) {
- assert rev > 0 : "Revision must be positive.";
-
- validate(type());
-
- type(ConditionType.REV_LESS);
-
- this.rev = rev;
-
- return new SimpleCondition(this);
+ return new RevisionCondition(key, ConditionType.REV_LESS, rev);
}
/**
@@ -168,40 +175,48 @@ public final class SimpleCondition implements Condition {
*
* @param rev The revision.
* @return The condition of type {@link ConditionType#REV_LESS_OR_EQUAL}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition le(long rev) {
- assert rev > 0 : "Revision must be positive.";
-
- validate(type());
-
- type(ConditionType.REV_LESS_OR_EQUAL);
-
- this.rev = rev;
-
- return new SimpleCondition(this);
+ return new RevisionCondition(key, ConditionType.REV_LESS_OR_EQUAL, rev);
}
}
/**
- * Represents a condition on an entry value. Only the one type of condition could be applied to the one instance of a condition.
- * Subsequent invocations of any method, which produces a condition will throw {@link IllegalStateException}.
+ * Represents a condition on an entry revision.
*/
- public static final class ValueCondition extends AbstractCondition {
- /** The value as the condition argument. */
- private byte[] val;
+ public static final class RevisionCondition extends SimpleCondition {
+ /** The revision as the condition argument. */
+ private final long rev;
/**
- * Constructs a condition by a value for an entry identified by the given key.
+ * Constructs a condition by a revision for an entry identified by the given key.
*
* @param key Identifies an entry, which condition will be applied to.
*/
- ValueCondition(byte[] key) {
- super(key);
+ RevisionCondition(byte[] key, ConditionType type, long rev) {
+ super(key, type);
+
+ assert rev > 0 : "Revision must be positive";
+
+ this.rev = rev;
}
- public byte[] value() {
- return val;
+ /**
+ * Returns the revision that will be used as a part of this condition.
+ */
+ public long revision() {
+ return rev;
+ }
+ }
+
+ /**
+ * Builder for {@link ValueCondition}.
+ */
+ public static final class ValueConditionBuilder {
+ private final byte[] key;
+
+ ValueConditionBuilder(byte[] key) {
+ this.key = key;
}
/**
@@ -210,16 +225,9 @@ public final class SimpleCondition implements Condition {
*
* @param val The value.
* @return The condition of type {@link ConditionType#VAL_EQUAL}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition eq(byte[] val) {
- validate(type());
-
- type(ConditionType.VAL_EQUAL);
-
- this.val = val;
-
- return new SimpleCondition(this);
+ return new ValueCondition(key, ConditionType.VAL_EQUAL, val);
}
/**
@@ -228,35 +236,20 @@ public final class SimpleCondition implements Condition {
*
* @param val The value.
* @return The condition of type {@link ConditionType#VAL_NOT_EQUAL}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition ne(byte[] val) {
- validate(type());
-
- type(ConditionType.VAL_NOT_EQUAL);
-
- this.val = val;
-
- return new SimpleCondition(this);
+ return new ValueCondition(key, ConditionType.VAL_NOT_EQUAL, val);
}
-
/**
* Produces the condition of type {@link ConditionType#VAL_GREATER}. This condition tests that the target entry value is greater
* than the given value in the lexicographical order.
*
* @param val The value.
* @return The condition of type {@link ConditionType#VAL_GREATER}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition gt(byte[] val) {
- validate(type());
-
- type(ConditionType.VAL_GREATER);
-
- this.val = val;
-
- return new SimpleCondition(this);
+ return new ValueCondition(key, ConditionType.VAL_GREATER, val);
}
/**
@@ -265,16 +258,9 @@ public final class SimpleCondition implements Condition {
*
* @param val The value.
* @return The condition of type {@link ConditionType#VAL_GREATER_OR_EQUAL}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition ge(byte[] val) {
- validate(type());
-
- type(ConditionType.VAL_GREATER_OR_EQUAL);
-
- this.val = val;
-
- return new SimpleCondition(this);
+ return new ValueCondition(key, ConditionType.VAL_GREATER_OR_EQUAL, val);
}
/**
@@ -283,16 +269,9 @@ public final class SimpleCondition implements Condition {
*
* @param val The value.
* @return The condition of type {@link ConditionType#VAL_LESS}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition lt(byte[] val) {
- validate(type());
-
- type(ConditionType.VAL_LESS);
-
- this.val = val;
-
- return new SimpleCondition(this);
+ return new ValueCondition(key, ConditionType.VAL_LESS, val);
}
/**
@@ -301,158 +280,35 @@ public final class SimpleCondition implements Condition {
*
* @param val The value.
* @return The condition of type {@link ConditionType#VAL_LESS_OR_EQUAL}.
- * @throws IllegalStateException In the case when the condition is already defined.
*/
public SimpleCondition le(byte[] val) {
- validate(type());
-
- type(ConditionType.VAL_LESS_OR_EQUAL);
-
- this.val = val;
-
- return new SimpleCondition(this);
+ return new ValueCondition(key, ConditionType.VAL_LESS_OR_EQUAL, val);
}
}
/**
- * Represents a condition on an entry existence. Only the one type of a condition could be applied to the one instance of a condition.
- * Subsequent invocations of any method, which produces a condition will throw {@link IllegalStateException}.
+ * Represents a condition on an entry value.
*/
- public static final class ExistenceCondition extends AbstractCondition {
- /**
- * Constructs a condition on existence an entry identified by the given key.
- *
- * @param key Identifies an entry, which condition will be applied to.
- */
- ExistenceCondition(byte[] key) {
- super(key);
- }
-
- /**
- * Produces the condition of type {@link ConditionType#KEY_EXISTS}. This condition tests the existence of an entry identified by the
- * given key.
- *
- * @return The condition of type {@link ConditionType#KEY_EXISTS}.
- * @throws IllegalStateException In the case when the condition is already defined.
- */
- public SimpleCondition exists() {
- validate(type());
-
- type(ConditionType.KEY_EXISTS);
-
- return new SimpleCondition(this);
- }
-
- /**
- * Produces the condition of type {@link ConditionType#KEY_NOT_EXISTS}. This condition tests the non-existence of an entry
- * identified by the given key.
- *
- * @return The condition of type {@link ConditionType#KEY_NOT_EXISTS}.
- * @throws IllegalStateException In the case when the condition is already defined.
- */
- public SimpleCondition notExists() {
- validate(type());
-
- type(ConditionType.KEY_NOT_EXISTS);
-
- return new SimpleCondition(this);
- }
- }
+ public static final class ValueCondition extends SimpleCondition {
+ /** The value as the condition argument. */
+ private final byte[] val;
- /**
- * Represents a condition on an entry's value, which checks whether a value is tombstone or not. Only the one type of condition
- * could be applied to the one instance of a condition. Subsequent invocations of any method which produces a condition
- * will throw {@link IllegalStateException}.
- */
- public static final class TombstoneCondition extends AbstractCondition {
/**
- * Constructs a condition on an entry, identified by the given key, is tombstone.
+ * Constructs a condition by a value for an entry identified by the given key.
*
* @param key Identifies an entry, which condition will be applied to.
*/
- TombstoneCondition(byte[] key) {
- super(key);
- }
-
- /**
- * Produces the condition of type {@link ConditionType#TOMBSTONE}. This condition tests that an entry's value, identified by the
- * given key, is tombstone.
- *
- * @return The condition of type {@link ConditionType#TOMBSTONE}.
- * @throws IllegalStateException In the case when the condition is already defined.
- */
- public SimpleCondition tombstone() {
- validate(type());
-
- type(ConditionType.TOMBSTONE);
-
- return new SimpleCondition(this);
- }
- }
-
- /**
- * Checks that condition is not defined yet. If the condition is already defined then the {@link IllegalStateException} will be thrown.
- *
- * @throws IllegalStateException In the case when the condition is already defined.
- */
- private static void validate(Enum<?> type) {
- if (type != null) {
- throw new IllegalStateException("Condition type " + type.name() + " is already defined.");
- }
- }
-
- /**
- * Defines a condition interface.
- */
- public interface InnerCondition {
- /**
- * Returns a key, which identifies an entry, which condition will be applied to.
- *
- * @return Key, which identifies an entry, which condition will be applied to.
- */
- byte[] key();
-
- ConditionType type();
- }
-
- /**
- * Defines an abstract condition with the key, which identifies an entry, which condition will be applied to.
- */
- private abstract static class AbstractCondition implements InnerCondition {
- /** Entry key. */
- private final byte[] key;
-
- /**
- * Condition type.
- */
- private ConditionType type;
+ ValueCondition(byte[] key, ConditionType type, byte[] val) {
+ super(key, type);
- /**
- * Constructs a condition with the given entry key.
- *
- * @param key Key, which identifies an entry, which condition will be applied to.
- */
- private AbstractCondition(byte[] key) {
- this.key = key;
+ this.val = val;
}
/**
- * Returns the key, which identifies an entry, which condition will be applied to.
- *
- * @return Key, which identifies an entry, which condition will be applied to.
+ * Returns the value that will be used as a part of this condition.
*/
- @Override
- public byte[] key() {
- return key;
- }
-
- @Override
- public ConditionType type() {
- return type;
- }
-
- protected void type(ConditionType type) {
- this.type = type;
+ public byte[] value() {
+ return val;
}
}
}
diff --git a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageServicePersistenceTest.java b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageServicePersistenceTest.java
index 96c30b2265..c8f0ad554f 100644
--- a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageServicePersistenceTest.java
+++ b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageServicePersistenceTest.java
@@ -21,7 +21,6 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
-import java.util.Arrays;
import java.util.concurrent.ExecutionException;
import java.util.function.BooleanSupplier;
import org.apache.ignite.internal.metastorage.Entry;
@@ -69,7 +68,7 @@ public class ItMetaStorageServicePersistenceTest extends ItAbstractListenerSnaps
metaStorage.put(FIRST_KEY, FIRST_VALUE).get();
// Check that data has been written successfully
- check(metaStorage, new EntryImpl(FIRST_KEY, FIRST_VALUE, 1, 1));
+ check(metaStorage, new EntryImpl(FIRST_KEY.bytes(), FIRST_VALUE, 1, 1));
}
/** {@inheritDoc} */
@@ -79,13 +78,13 @@ public class ItMetaStorageServicePersistenceTest extends ItAbstractListenerSnaps
metaStorage.remove(FIRST_KEY).get();
// Check that data has been removed
- check(metaStorage, new EntryImpl(FIRST_KEY, null, 2, 2));
+ check(metaStorage, new EntryImpl(FIRST_KEY.bytes(), null, 2, 2));
// Put same data again
metaStorage.put(FIRST_KEY, FIRST_VALUE).get();
// Check that it has been written
- check(metaStorage, new EntryImpl(FIRST_KEY, FIRST_VALUE, 3, 3));
+ check(metaStorage, new EntryImpl(FIRST_KEY.bytes(), FIRST_VALUE, 3, 3));
}
/** {@inheritDoc} */
@@ -105,17 +104,9 @@ public class ItMetaStorageServicePersistenceTest extends ItAbstractListenerSnaps
int expectedRevision = interactedAfterSnapshot ? 4 : 3;
int expectedUpdateCounter = interactedAfterSnapshot ? 4 : 3;
- EntryImpl expectedLastEntry = new EntryImpl(new ByteArray(lastKey), lastValue, expectedRevision, expectedUpdateCounter);
-
- return () -> {
- org.apache.ignite.internal.metastorage.server.Entry e = storage.get(lastKey);
- return e.empty() == expectedLastEntry.empty()
- && e.tombstone() == expectedLastEntry.tombstone()
- && e.revision() == expectedLastEntry.revision()
- && e.updateCounter() == expectedLastEntry.revision()
- && Arrays.equals(e.key(), expectedLastEntry.key().bytes())
- && Arrays.equals(e.value(), expectedLastEntry.value());
- };
+ EntryImpl expectedLastEntry = new EntryImpl(lastKey, lastValue, expectedRevision, expectedUpdateCounter);
+
+ return () -> storage.get(lastKey).equals(expectedLastEntry);
}
/** {@inheritDoc} */
@@ -150,7 +141,7 @@ public class ItMetaStorageServicePersistenceTest extends ItAbstractListenerSnaps
*/
private static void check(MetaStorageServiceImpl metaStorage, EntryImpl expected)
throws ExecutionException, InterruptedException {
- Entry entry = metaStorage.get(expected.key()).get();
+ Entry entry = metaStorage.get(new ByteArray(expected.key())).get();
assertEquals(expected, entry);
}
diff --git a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageServiceTest.java b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageServiceTest.java
index d43355d4af..ef21661168 100644
--- a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageServiceTest.java
+++ b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageServiceTest.java
@@ -71,6 +71,7 @@ import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.EntryEvent;
import org.apache.ignite.internal.metastorage.WatchEvent;
import org.apache.ignite.internal.metastorage.WatchListener;
import org.apache.ignite.internal.metastorage.dsl.Condition;
@@ -79,17 +80,16 @@ import org.apache.ignite.internal.metastorage.dsl.If;
import org.apache.ignite.internal.metastorage.dsl.Operation;
import org.apache.ignite.internal.metastorage.dsl.OperationType;
import org.apache.ignite.internal.metastorage.dsl.Operations;
+import org.apache.ignite.internal.metastorage.dsl.StatementResult;
+import org.apache.ignite.internal.metastorage.dsl.Update;
import org.apache.ignite.internal.metastorage.exceptions.CompactedException;
import org.apache.ignite.internal.metastorage.exceptions.OperationTimeoutException;
import org.apache.ignite.internal.metastorage.server.AbstractCompoundCondition;
import org.apache.ignite.internal.metastorage.server.AbstractSimpleCondition;
import org.apache.ignite.internal.metastorage.server.AndCondition;
-import org.apache.ignite.internal.metastorage.server.EntryEvent;
import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
import org.apache.ignite.internal.metastorage.server.OrCondition;
import org.apache.ignite.internal.metastorage.server.RevisionCondition;
-import org.apache.ignite.internal.metastorage.server.StatementResult;
-import org.apache.ignite.internal.metastorage.server.Update;
import org.apache.ignite.internal.metastorage.server.ValueCondition;
import org.apache.ignite.internal.metastorage.server.ValueCondition.Type;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageListener;
@@ -142,21 +142,12 @@ public class ItMetaStorageServiceTest {
/** Nodes. */
private static final int NODES = 2;
- /** Expected server result entry. */
- private static final org.apache.ignite.internal.metastorage.server.Entry EXPECTED_SRV_RESULT_ENTRY =
- new org.apache.ignite.internal.metastorage.server.Entry(
- new byte[]{1},
- new byte[]{2},
- 10,
- 2
- );
-
/**
* Expected server result entry.
*/
private static final EntryImpl EXPECTED_RESULT_ENTRY =
new EntryImpl(
- new ByteArray(new byte[]{1}),
+ new byte[]{1},
new byte[]{2},
10,
2
@@ -168,7 +159,7 @@ public class ItMetaStorageServiceTest {
private static final NavigableMap<ByteArray, Entry> EXPECTED_RESULT_MAP;
/** Expected server result collection. */
- private static final Collection<org.apache.ignite.internal.metastorage.server.Entry> EXPECTED_SRV_RESULT_COLL;
+ private static final Collection<Entry> EXPECTED_SRV_RESULT_COLL;
/** Node 0 id. */
private static final String NODE_ID_0 = "node-id-0";
@@ -202,31 +193,24 @@ public class ItMetaStorageServiceTest {
EXPECTED_RESULT_MAP = new TreeMap<>();
EntryImpl entry1 = new EntryImpl(
- new ByteArray(new byte[]{1}),
+ new byte[]{1},
new byte[]{2},
10,
2
);
- EXPECTED_RESULT_MAP.put(entry1.key(), entry1);
+ EXPECTED_RESULT_MAP.put(new ByteArray(entry1.key()), entry1);
EntryImpl entry2 = new EntryImpl(
- new ByteArray(new byte[]{3}),
+ new byte[]{3},
new byte[]{4},
10,
3
);
- EXPECTED_RESULT_MAP.put(entry2.key(), entry2);
+ EXPECTED_RESULT_MAP.put(new ByteArray(entry2.key()), entry2);
- EXPECTED_SRV_RESULT_COLL = List.of(
- new org.apache.ignite.internal.metastorage.server.Entry(
- entry1.key().bytes(), entry1.value(), entry1.revision(), entry1.updateCounter()
- ),
- new org.apache.ignite.internal.metastorage.server.Entry(
- entry2.key().bytes(), entry2.value(), entry2.revision(), entry2.updateCounter()
- )
- );
+ EXPECTED_SRV_RESULT_COLL = List.of(entry1, entry2);
}
/**
@@ -280,9 +264,9 @@ public class ItMetaStorageServiceTest {
*/
@Test
public void testGet() throws Exception {
- when(mockStorage.get(EXPECTED_RESULT_ENTRY.key().bytes())).thenReturn(EXPECTED_SRV_RESULT_ENTRY);
+ when(mockStorage.get(EXPECTED_RESULT_ENTRY.key())).thenReturn(EXPECTED_RESULT_ENTRY);
- assertEquals(EXPECTED_RESULT_ENTRY, metaStorageSvc.get(EXPECTED_RESULT_ENTRY.key()).get());
+ assertEquals(EXPECTED_RESULT_ENTRY, metaStorageSvc.get(new ByteArray(EXPECTED_RESULT_ENTRY.key())).get());
}
/**
@@ -292,12 +276,11 @@ public class ItMetaStorageServiceTest {
*/
@Test
public void testGetWithUpperBoundRevision() throws Exception {
- when(mockStorage.get(EXPECTED_RESULT_ENTRY.key().bytes(), EXPECTED_RESULT_ENTRY.revision()))
- .thenReturn(EXPECTED_SRV_RESULT_ENTRY);
+ when(mockStorage.get(EXPECTED_RESULT_ENTRY.key(), EXPECTED_RESULT_ENTRY.revision())).thenReturn(EXPECTED_RESULT_ENTRY);
assertEquals(
EXPECTED_RESULT_ENTRY,
- metaStorageSvc.get(EXPECTED_RESULT_ENTRY.key(), EXPECTED_RESULT_ENTRY.revision()).get()
+ metaStorageSvc.get(new ByteArray(EXPECTED_RESULT_ENTRY.key()), EXPECTED_RESULT_ENTRY.revision()).get()
);
}
@@ -353,11 +336,11 @@ public class ItMetaStorageServiceTest {
public void testGetAndPut() throws Exception {
byte[] expVal = {2};
- when(mockStorage.getAndPut(EXPECTED_RESULT_ENTRY.key().bytes(), expVal)).thenReturn(EXPECTED_SRV_RESULT_ENTRY);
+ when(mockStorage.getAndPut(EXPECTED_RESULT_ENTRY.key(), expVal)).thenReturn(EXPECTED_RESULT_ENTRY);
assertEquals(
EXPECTED_RESULT_ENTRY,
- metaStorageSvc.getAndPut(EXPECTED_RESULT_ENTRY.key(), expVal).get()
+ metaStorageSvc.getAndPut(new ByteArray(EXPECTED_RESULT_ENTRY.key()), expVal).get()
);
}
@@ -469,22 +452,15 @@ public class ItMetaStorageServiceTest {
@Test
public void testGetAndRemove() throws Exception {
EntryImpl expRes = new EntryImpl(
- new ByteArray(new byte[]{1}),
+ new byte[]{1},
new byte[]{3},
10,
2
);
- when(mockStorage.getAndRemove(expRes.key().bytes())).thenReturn(
- new org.apache.ignite.internal.metastorage.server.Entry(
- expRes.key().bytes(),
- expRes.value(),
- expRes.revision(),
- expRes.updateCounter()
- )
- );
+ when(mockStorage.getAndRemove(expRes.key())).thenReturn(expRes);
- assertEquals(expRes, metaStorageSvc.getAndRemove(expRes.key()).get());
+ assertEquals(expRes, metaStorageSvc.getAndRemove(new ByteArray(expRes.key())).get());
}
/**
@@ -543,10 +519,9 @@ public class ItMetaStorageServiceTest {
/**
* Tests {@link MetaStorageService#range(ByteArray, ByteArray, long)}} with not null keyTo and explicit revUpperBound.
*
- * @throws Exception If failed.
*/
@Test
- public void testRangeWitKeyToAndUpperBound() throws Exception {
+ public void testRangeWitKeyToAndUpperBound() {
ByteArray expKeyFrom = new ByteArray(new byte[]{1});
ByteArray expKeyTo = new ByteArray(new byte[]{3});
@@ -561,10 +536,9 @@ public class ItMetaStorageServiceTest {
/**
* Tests {@link MetaStorageService#range(ByteArray, ByteArray, long)}} with not null keyTo.
*
- * @throws Exception If failed.
*/
@Test
- public void testRangeWitKeyTo() throws Exception {
+ public void testRangeWitKeyTo() {
ByteArray expKeyFrom = new ByteArray(new byte[]{1});
ByteArray expKeyTo = new ByteArray(new byte[]{3});
@@ -577,10 +551,9 @@ public class ItMetaStorageServiceTest {
/**
* Tests {@link MetaStorageService#range(ByteArray, ByteArray, long)}} with null keyTo.
*
- * @throws Exception If failed.
*/
@Test
- public void testRangeWitNullAsKeyTo() throws Exception {
+ public void testRangeWitNullAsKeyTo() {
ByteArray expKeyFrom = new ByteArray(new byte[]{1});
when(mockStorage.range(expKeyFrom.bytes(), null, false)).thenReturn(mock(Cursor.class));
@@ -613,16 +586,16 @@ public class ItMetaStorageServiceTest {
*/
@Test
public void testRangeNext() {
- when(mockStorage.range(EXPECTED_RESULT_ENTRY.key().bytes(), null, false)).thenAnswer(invocation -> {
+ when(mockStorage.range(EXPECTED_RESULT_ENTRY.key(), null, false)).thenAnswer(invocation -> {
var cursor = mock(Cursor.class);
when(cursor.hasNext()).thenReturn(true);
- when(cursor.next()).thenReturn(EXPECTED_SRV_RESULT_ENTRY);
+ when(cursor.next()).thenReturn(EXPECTED_RESULT_ENTRY);
return cursor;
});
- Cursor<Entry> cursor = metaStorageSvc.range(EXPECTED_RESULT_ENTRY.key(), null);
+ Cursor<Entry> cursor = metaStorageSvc.range(new ByteArray(EXPECTED_RESULT_ENTRY.key()), null);
assertEquals(EXPECTED_RESULT_ENTRY, cursor.iterator().next());
}
@@ -632,7 +605,7 @@ public class ItMetaStorageServiceTest {
*/
@Test
public void testRangeNextNoSuchElementException() {
- when(mockStorage.range(EXPECTED_RESULT_ENTRY.key().bytes(), null, false)).thenAnswer(invocation -> {
+ when(mockStorage.range(EXPECTED_RESULT_ENTRY.key(), null, false)).thenAnswer(invocation -> {
var cursor = mock(Cursor.class);
when(cursor.hasNext()).thenReturn(true);
@@ -641,7 +614,7 @@ public class ItMetaStorageServiceTest {
return cursor;
});
- Cursor<Entry> cursor = metaStorageSvc.range(EXPECTED_RESULT_ENTRY.key(), null);
+ Cursor<Entry> cursor = metaStorageSvc.range(new ByteArray(EXPECTED_RESULT_ENTRY.key()), null);
assertThrows(NoSuchElementException.class, () -> cursor.iterator().next());
}
@@ -649,10 +622,9 @@ public class ItMetaStorageServiceTest {
/**
* Tests {@link MetaStorageService#range(ByteArray, ByteArray, long)}} close.
*
- * @throws Exception If failed.
*/
@Test
- public void testRangeClose() throws Exception {
+ public void testRangeClose() {
ByteArray expKeyFrom = new ByteArray(new byte[]{1});
Cursor cursorMock = mock(Cursor.class);
@@ -668,44 +640,44 @@ public class ItMetaStorageServiceTest {
@Test
public void testWatchOnUpdate() throws Exception {
- org.apache.ignite.internal.metastorage.server.WatchEvent expectedEvent =
- new org.apache.ignite.internal.metastorage.server.WatchEvent(List.of(
- new org.apache.ignite.internal.metastorage.server.EntryEvent(
- new org.apache.ignite.internal.metastorage.server.Entry(
+ WatchEvent expectedEvent =
+ new WatchEvent(List.of(
+ new EntryEvent(
+ new EntryImpl(
new byte[]{2},
new byte[]{20},
1,
1
),
- new org.apache.ignite.internal.metastorage.server.Entry(
+ new EntryImpl(
new byte[]{2},
new byte[]{21},
2,
4
)
),
- new org.apache.ignite.internal.metastorage.server.EntryEvent(
- new org.apache.ignite.internal.metastorage.server.Entry(
+ new EntryEvent(
+ new EntryImpl(
new byte[]{3},
new byte[]{20},
1,
2
),
- new org.apache.ignite.internal.metastorage.server.Entry(
+ new EntryImpl(
new byte[]{3},
new byte[]{},
2,
5
)
),
- new org.apache.ignite.internal.metastorage.server.EntryEvent(
- new org.apache.ignite.internal.metastorage.server.Entry(
+ new EntryEvent(
+ new EntryImpl(
new byte[]{4},
new byte[]{20},
1,
3
),
- new org.apache.ignite.internal.metastorage.server.Entry(
+ new EntryImpl(
new byte[]{4},
new byte[]{},
3,
@@ -735,21 +707,21 @@ public class ItMetaStorageServiceTest {
@Override
public boolean onUpdate(@NotNull WatchEvent event) {
Collection<EntryEvent> expectedEvents = expectedEvent.entryEvents();
- Collection<org.apache.ignite.internal.metastorage.EntryEvent> actualEvents = event.entryEvents();
+ Collection<EntryEvent> actualEvents = event.entryEvents();
assertEquals(expectedEvents.size(), actualEvents.size());
Iterator<EntryEvent> expectedIterator = expectedEvents.iterator();
- Iterator<org.apache.ignite.internal.metastorage.EntryEvent> actualIterator = actualEvents.iterator();
+ Iterator<EntryEvent> actualIterator = actualEvents.iterator();
while (expectedIterator.hasNext() && actualIterator.hasNext()) {
- org.apache.ignite.internal.metastorage.server.EntryEvent expectedEntryEvent = expectedIterator.next();
- org.apache.ignite.internal.metastorage.EntryEvent actualEntryEvent = actualIterator.next();
+ EntryEvent expectedEntryEvent = expectedIterator.next();
+ EntryEvent actualEntryEvent = actualIterator.next();
- assertArrayEquals(expectedEntryEvent.oldEntry().key(), actualEntryEvent.oldEntry().key().bytes());
+ assertArrayEquals(expectedEntryEvent.oldEntry().key(), actualEntryEvent.oldEntry().key());
assertArrayEquals(expectedEntryEvent.oldEntry().value(), actualEntryEvent.oldEntry().value());
- assertArrayEquals(expectedEntryEvent.entry().key(), actualEntryEvent.newEntry().key().bytes());
- assertArrayEquals(expectedEntryEvent.entry().value(), actualEntryEvent.newEntry().value());
+ assertArrayEquals(expectedEntryEvent.newEntry().key(), actualEntryEvent.newEntry().key());
+ assertArrayEquals(expectedEntryEvent.newEntry().value(), actualEntryEvent.newEntry().value());
}
latch.countDown();
@@ -829,19 +801,18 @@ public class ItMetaStorageServiceTest {
Type.LESS_OR_EQUAL, key1.bytes(), val2))));
assertThat(resultIf.andThen().iif().andThen().update(), upd(new Update(
- List.of(new org.apache.ignite.internal.metastorage.server.Operation(OperationType.PUT, key1.bytes(), rval1)),
+ List.of(Operation.put(key1.bytes(), rval1)),
new StatementResult(true))));
assertThat(resultIf.andThen().iif().orElse().iif().andThen().update(), upd(new Update(
- Arrays.asList(new org.apache.ignite.internal.metastorage.server.Operation(OperationType.PUT, key1.bytes(), rval1),
- new org.apache.ignite.internal.metastorage.server.Operation(OperationType.REMOVE, key2.bytes(), null)),
+ Arrays.asList(Operation.put(key1.bytes(), rval1), Operation.remove(key2.bytes())),
new StatementResult(false))));
assertThat(resultIf.andThen().iif().orElse().iif().orElse().update(),
upd(new Update(Collections.emptyList(), new StatementResult(true))));
assertThat(resultIf.orElse().update(), upd(new Update(
- List.of(new org.apache.ignite.internal.metastorage.server.Operation(OperationType.PUT, key2.bytes(), rval2)),
+ List.of(Operation.put(key2.bytes(), rval2)),
new StatementResult(false))));
}
@@ -863,11 +834,9 @@ public class ItMetaStorageServiceTest {
var conditionCaptor = ArgumentCaptor.forClass(AbstractSimpleCondition.class);
- ArgumentCaptor<Collection<org.apache.ignite.internal.metastorage.server.Operation>> successCaptor =
- ArgumentCaptor.forClass(Collection.class);
+ ArgumentCaptor<Collection<Operation>> successCaptor = ArgumentCaptor.forClass(Collection.class);
- ArgumentCaptor<Collection<org.apache.ignite.internal.metastorage.server.Operation>> failureCaptor =
- ArgumentCaptor.forClass(Collection.class);
+ ArgumentCaptor<Collection<Operation>> failureCaptor = ArgumentCaptor.forClass(Collection.class);
verify(mockStorage).invoke(conditionCaptor.capture(), successCaptor.capture(), failureCaptor.capture());
@@ -888,10 +857,10 @@ public class ItMetaStorageServiceTest {
@Disabled // TODO: IGNITE-14693 Add tests for exception handling logic.
@Test
public void testGetThatThrowsCompactedException() {
- when(mockStorage.get(EXPECTED_RESULT_ENTRY.key().bytes()))
+ when(mockStorage.get(EXPECTED_RESULT_ENTRY.key()))
.thenThrow(new org.apache.ignite.internal.metastorage.server.CompactedException());
- assertThrows(CompactedException.class, () -> metaStorageSvc.get(EXPECTED_RESULT_ENTRY.key()).get());
+ assertThrows(CompactedException.class, () -> metaStorageSvc.get(new ByteArray(EXPECTED_RESULT_ENTRY.key())).get());
}
/**
@@ -900,9 +869,9 @@ public class ItMetaStorageServiceTest {
@Disabled // TODO: IGNITE-14693 Add tests for exception handling logic.
@Test
public void testGetThatThrowsOperationTimeoutException() {
- when(mockStorage.get(EXPECTED_RESULT_ENTRY.key().bytes())).thenThrow(new OperationTimeoutException());
+ when(mockStorage.get(EXPECTED_RESULT_ENTRY.key())).thenThrow(new OperationTimeoutException());
- assertThrows(OperationTimeoutException.class, () -> metaStorageSvc.get(EXPECTED_RESULT_ENTRY.key()).get());
+ assertThrows(OperationTimeoutException.class, () -> metaStorageSvc.get(new ByteArray(EXPECTED_RESULT_ENTRY.key())).get());
}
/**
@@ -912,11 +881,11 @@ public class ItMetaStorageServiceTest {
*/
@Test
public void testCursorsCleanup() throws Exception {
- when(mockStorage.range(EXPECTED_RESULT_ENTRY.key().bytes(), null, false)).thenAnswer(invocation -> {
+ when(mockStorage.range(EXPECTED_RESULT_ENTRY.key(), null, false)).thenAnswer(invocation -> {
var cursor = mock(Cursor.class);
when(cursor.hasNext()).thenReturn(true);
- when(cursor.next()).thenReturn(EXPECTED_SRV_RESULT_ENTRY);
+ when(cursor.next()).thenReturn(EXPECTED_RESULT_ENTRY);
return cursor;
});
@@ -933,15 +902,15 @@ public class ItMetaStorageServiceTest {
MetaStorageService metaStorageSvc2 = new MetaStorageServiceImpl(metaStorageRaftSvc2, NODE_ID_1, NODE_ID_1);
- Cursor<Entry> cursorNode0 = metaStorageSvc.range(EXPECTED_RESULT_ENTRY.key(), null);
+ Cursor<Entry> cursorNode0 = metaStorageSvc.range(new ByteArray(EXPECTED_RESULT_ENTRY.key()), null);
assertTrue(cursorNode0.hasNext());
- Cursor<Entry> cursor2Node0 = metaStorageSvc.range(EXPECTED_RESULT_ENTRY.key(), null);
+ Cursor<Entry> cursor2Node0 = metaStorageSvc.range(new ByteArray(EXPECTED_RESULT_ENTRY.key()), null);
assertTrue(cursor2Node0.hasNext());
- Cursor<Entry> cursorNode1 = metaStorageSvc2.range(EXPECTED_RESULT_ENTRY.key(), null);
+ Cursor<Entry> cursorNode1 = metaStorageSvc2.range(new ByteArray(EXPECTED_RESULT_ENTRY.key()), null);
assertTrue(cursorNode1.hasNext());
@@ -993,8 +962,7 @@ public class ItMetaStorageServiceTest {
.collect(Collectors.joining(",")) + "])";
}
- private boolean opsEqual(Iterator<org.apache.ignite.internal.metastorage.server.Operation> ops1,
- Iterator<org.apache.ignite.internal.metastorage.server.Operation> ops2) {
+ private boolean opsEqual(Iterator<Operation> ops1, Iterator<Operation> ops2) {
if (!ops1.hasNext()) {
return true;
} else {
@@ -1002,8 +970,7 @@ public class ItMetaStorageServiceTest {
}
}
- private boolean opEqual(org.apache.ignite.internal.metastorage.server.Operation op1,
- org.apache.ignite.internal.metastorage.server.Operation op2) {
+ private boolean opEqual(Operation op1, Operation op2) {
return Arrays.equals(op1.key(), op2.key()) && Arrays.equals(op1.value(), op2.value()) && op1.type() == op2.type();
}
}
@@ -1013,7 +980,7 @@ public class ItMetaStorageServiceTest {
*/
protected static class ServerConditionMatcher extends TypeSafeMatcher<org.apache.ignite.internal.metastorage.server.Condition> {
- private org.apache.ignite.internal.metastorage.server.Condition condition;
+ private final org.apache.ignite.internal.metastorage.server.Condition condition;
public ServerConditionMatcher(org.apache.ignite.internal.metastorage.server.Condition condition) {
this.condition = condition;
diff --git a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/server/raft/ItMetaStorageRaftGroupTest.java b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/server/raft/ItMetaStorageRaftGroupTest.java
index cb5965a619..af270fa3e7 100644
--- a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/server/raft/ItMetaStorageRaftGroupTest.java
+++ b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/server/raft/ItMetaStorageRaftGroupTest.java
@@ -95,28 +95,10 @@ public class ItMetaStorageRaftGroupTest {
/** Factory. */
private static final RaftMessagesFactory FACTORY = new RaftMessagesFactory();
- /** Expected server result entry. */
- private static final org.apache.ignite.internal.metastorage.server.Entry EXPECTED_SRV_RESULT_ENTRY1 =
- new org.apache.ignite.internal.metastorage.server.Entry(
- new byte[] {1},
- new byte[] {2},
- 10,
- 2
- );
-
- /** Expected server result entry. */
- private static final org.apache.ignite.internal.metastorage.server.Entry EXPECTED_SRV_RESULT_ENTRY2 =
- new org.apache.ignite.internal.metastorage.server.Entry(
- new byte[] {3},
- new byte[] {4},
- 11,
- 3
- );
-
/** Expected server result entry. */
private static final EntryImpl EXPECTED_RESULT_ENTRY1 =
new EntryImpl(
- new ByteArray(new byte[] {1}),
+ new byte[] {1},
new byte[] {2},
10,
2
@@ -125,7 +107,7 @@ public class ItMetaStorageRaftGroupTest {
/** Expected server result entry. */
private static final EntryImpl EXPECTED_RESULT_ENTRY2 =
new EntryImpl(
- new ByteArray(new byte[] {3}),
+ new byte[] {3},
new byte[] {4},
11,
3
@@ -232,9 +214,9 @@ public class ItMetaStorageRaftGroupTest {
final AtomicInteger replicatorStoppedCounter = new AtomicInteger(0);
- when(mockStorage.range(EXPECTED_RESULT_ENTRY1.key().bytes(), new byte[]{4}, false)).thenAnswer(invocation -> {
- List<org.apache.ignite.internal.metastorage.server.Entry> entries = new ArrayList<>(
- List.of(EXPECTED_SRV_RESULT_ENTRY1, EXPECTED_SRV_RESULT_ENTRY2));
+ when(mockStorage.range(EXPECTED_RESULT_ENTRY1.key(), new byte[]{4}, false)).thenAnswer(invocation -> {
+ List<Entry> entries = new ArrayList<>(
+ List.of(EXPECTED_RESULT_ENTRY1, EXPECTED_RESULT_ENTRY2));
return Cursor.fromBareIterator(entries.iterator());
});
@@ -264,7 +246,7 @@ public class ItMetaStorageRaftGroupTest {
MetaStorageService metaStorageSvc = new MetaStorageServiceImpl(
raftGroupServiceOfLiveServer, "some_node", "some_node");
- Cursor<Entry> cursor = metaStorageSvc.range(EXPECTED_RESULT_ENTRY1.key(), new ByteArray(new byte[]{4}));
+ Cursor<Entry> cursor = metaStorageSvc.range(new ByteArray(EXPECTED_RESULT_ENTRY1.key()), new ByteArray(new byte[]{4}));
assertTrue(waitForCondition(
() -> replicatorStartedCounter.get() == 2, 5_000), replicatorStartedCounter.get() + "");
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/command/SingleEntryResponse.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/command/SingleEntryResponse.java
index b6a861e551..1fa4b59a30 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/command/SingleEntryResponse.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/command/SingleEntryResponse.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.metastorage.command;
import java.io.Serializable;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -26,12 +25,10 @@ import org.jetbrains.annotations.Nullable;
*/
public class SingleEntryResponse implements Serializable {
/** Key. */
- @NotNull
private final byte[] key;
/** Value. */
- @Nullable
- private final byte[] val;
+ private final byte @Nullable [] val;
/** Revision. */
private final long rev;
@@ -47,7 +44,7 @@ public class SingleEntryResponse implements Serializable {
* @param rev Revision number.
* @param updCntr Update counter.
*/
- public SingleEntryResponse(byte[] key, byte[] val, long rev, long updCntr) {
+ public SingleEntryResponse(byte[] key, byte @Nullable [] val, long rev, long updCntr) {
this.key = key;
this.val = val;
this.rev = rev;
@@ -59,7 +56,6 @@ public class SingleEntryResponse implements Serializable {
*
* @return Entry key. Couldn't be {@code null}.
*/
- @NotNull
public byte[] key() {
return key;
}
@@ -69,8 +65,7 @@ public class SingleEntryResponse implements Serializable {
*
* @return Entry value. Could be {@code null} for empty and tombstone entries.
*/
- @Nullable
- public byte[] value() {
+ public byte @Nullable [] value() {
return val;
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/EntryImpl.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/EntryImpl.java
index a48e147f8e..1b38283b52 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/EntryImpl.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/EntryImpl.java
@@ -19,17 +19,32 @@ package org.apache.ignite.internal.metastorage.impl;
import java.util.Arrays;
import org.apache.ignite.internal.metastorage.Entry;
-import org.apache.ignite.lang.ByteArray;
-import org.jetbrains.annotations.NotNull;
+import org.apache.ignite.internal.tostring.S;
import org.jetbrains.annotations.Nullable;
/**
- * Meta storage entry.
+ * Represents a storage unit as entry with key, value and revision.
+ *
+ * <p>Where:
+ * <ul>
+ * <li>key - an unique entry's key represented by an array of bytes. Keys are comparable in lexicographic manner.</li>
+ * <li>value - a data which is associated with a key and represented as an array of bytes.</li>
+ * <li>revision - a number which denotes a version of whole meta storage.
+ * Each change (which could include multiple entries) increments the revision. </li>
+ * <li>updateCounter - a number which increments on every update in the change under one revision.</li>
+ * </ul>
+ *
+ * <p>Instance of {@link #EntryImpl} could represent:
+ * <ul>
+ * <li>A regular entry which stores a particular key, a value and a revision number.</li>
+ * <li>An empty entry which denotes absence of a regular entry in the meta storage for a given key.
+ * A revision is 0 for such kind of entry.</li>
+ * <li>A tombstone entry which denotes that a regular entry for a given key was removed from the storage at some revision.</li>
+ * </ul>
*/
public final class EntryImpl implements Entry {
/** Key. */
- @NotNull
- private final ByteArray key;
+ private final byte[] key;
/** Value. */
private final byte @Nullable [] val;
@@ -41,14 +56,14 @@ public final class EntryImpl implements Entry {
private final long updCntr;
/**
- * Construct entry with given paramteters.
+ * Construct entry with given parameters.
*
- * @param key Key.
- * @param val Value.
- * @param rev Revision.
+ * @param key Key.
+ * @param val Value.
+ * @param rev Revision.
* @param updCntr Update counter.
*/
- public EntryImpl(@NotNull ByteArray key, byte @Nullable [] val, long rev, long updCntr) {
+ public EntryImpl(byte[] key, byte @Nullable [] val, long rev, long updCntr) {
this.key = key;
this.val = val;
this.rev = rev;
@@ -56,9 +71,8 @@ public final class EntryImpl implements Entry {
}
/** {@inheritDoc} */
- @NotNull
@Override
- public ByteArray key() {
+ public byte[] key() {
return key;
}
@@ -80,19 +94,40 @@ public final class EntryImpl implements Entry {
return updCntr;
}
+ /**
+ * Creates an instance of tombstone entry for a given key and a revision.
+ *
+ * @param key Key bytes. Couldn't be {@code null}.
+ * @param rev Revision.
+ * @param updCntr Update counter.
+ * @return Empty entry.
+ */
+ public static Entry tombstone(byte[] key, long rev, long updCntr) {
+ return new EntryImpl(key, null, rev, updCntr);
+ }
+
/** {@inheritDoc} */
@Override
public boolean tombstone() {
return val == null && rev > 0 && updCntr > 0;
}
+ /**
+ * Creates an instance of empty entry for a given key.
+ *
+ * @param key Key bytes. Couldn't be {@code null}.
+ * @return Empty entry.
+ */
+ public static Entry empty(byte[] key) {
+ return new EntryImpl(key, null, 0, 0);
+ }
+
/** {@inheritDoc} */
@Override
public boolean empty() {
return val == null && rev == 0 && updCntr == 0;
}
-
/** {@inheritDoc} */
@Override
public boolean equals(Object o) {
@@ -114,7 +149,7 @@ public final class EntryImpl implements Entry {
return false;
}
- if (!key.equals(entry.key)) {
+ if (!Arrays.equals(key, entry.key)) {
return false;
}
@@ -124,7 +159,7 @@ public final class EntryImpl implements Entry {
/** {@inheritDoc} */
@Override
public int hashCode() {
- int res = key.hashCode();
+ int res = Arrays.hashCode(key);
res = 31 * res + Arrays.hashCode(val);
@@ -134,4 +169,9 @@ public final class EntryImpl implements Entry {
return res;
}
+
+ @Override
+ public String toString() {
+ return S.toString(this);
+ }
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageService.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageService.java
index 46c0fed72b..8aa2ee96da 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageService.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageService.java
@@ -32,7 +32,6 @@ import org.apache.ignite.internal.metastorage.exceptions.OperationTimeoutExcepti
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteUuid;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -48,8 +47,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<Entry> get(@NotNull ByteArray key);
+ CompletableFuture<Entry> get(ByteArray key);
/**
* Retrieves an entry for the given key and the revision upper bound.
@@ -63,8 +61,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<Entry> get(@NotNull ByteArray key, long revUpperBound);
+ CompletableFuture<Entry> get(ByteArray key, long revUpperBound);
/**
* Retrieves entries for given keys.
@@ -75,7 +72,6 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
CompletableFuture<Map<ByteArray, Entry>> getAll(Set<ByteArray> keys);
/**
@@ -90,7 +86,6 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
CompletableFuture<Map<ByteArray, Entry>> getAll(Set<ByteArray> keys, long revUpperBound);
/**
@@ -103,8 +98,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<Void> put(@NotNull ByteArray key, @NotNull byte[] value);
+ CompletableFuture<Void> put(ByteArray key, byte[] value);
/**
* Inserts or updates an entry with the given key and the given value and retrieves a previous entry for the given key.
@@ -116,8 +110,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<Entry> getAndPut(@NotNull ByteArray key, @NotNull byte[] value);
+ CompletableFuture<Entry> getAndPut(ByteArray key, byte[] value);
/**
* Inserts or updates entries with given keys and given values.
@@ -128,8 +121,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<Void> putAll(@NotNull Map<ByteArray, byte[]> vals);
+ CompletableFuture<Void> putAll(Map<ByteArray, byte[]> vals);
/**
* Inserts or updates entries with given keys and given values and retrieves a previous entries for given keys.
@@ -140,8 +132,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<Map<ByteArray, Entry>> getAndPutAll(@NotNull Map<ByteArray, byte[]> vals);
+ CompletableFuture<Map<ByteArray, Entry>> getAndPutAll(Map<ByteArray, byte[]> vals);
/**
* Removes an entry for the given key.
@@ -152,8 +143,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<Void> remove(@NotNull ByteArray key);
+ CompletableFuture<Void> remove(ByteArray key);
/**
* Removes an entry for the given key.
@@ -164,8 +154,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<Entry> getAndRemove(@NotNull ByteArray key);
+ CompletableFuture<Entry> getAndRemove(ByteArray key);
/**
* Removes entries for given keys.
@@ -176,8 +165,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<Void> removeAll(@NotNull Set<ByteArray> keys);
+ CompletableFuture<Void> removeAll(Set<ByteArray> keys);
/**
* Removes entries for given keys and retrieves previous entries.
@@ -189,8 +177,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<Map<ByteArray, Entry>> getAndRemoveAll(@NotNull Set<ByteArray> keys);
+ CompletableFuture<Map<ByteArray, Entry>> getAndRemoveAll(Set<ByteArray> keys);
/**
* Updates an entry for the given key conditionally.
@@ -207,9 +194,7 @@ public interface MetaStorageService {
* @see Condition
* @see Operation
*/
- @NotNull
- CompletableFuture<Boolean> invoke(@NotNull Condition condition,
- @NotNull Operation success, @NotNull Operation failure);
+ CompletableFuture<Boolean> invoke(Condition condition, Operation success, Operation failure);
/**
* Updates an entry for the given key conditionally.
@@ -226,9 +211,7 @@ public interface MetaStorageService {
* @see Condition
* @see Operation
*/
- @NotNull
- CompletableFuture<Boolean> invoke(@NotNull Condition condition,
- @NotNull Collection<Operation> success, @NotNull Collection<Operation> failure);
+ CompletableFuture<Boolean> invoke(Condition condition, Collection<Operation> success, Collection<Operation> failure);
/**
* Invoke, which supports nested conditional statements.
@@ -240,8 +223,7 @@ public interface MetaStorageService {
* @see If
* @see StatementResult
*/
- @NotNull
- CompletableFuture<StatementResult> invoke(@NotNull If iif);
+ CompletableFuture<StatementResult> invoke(If iif);
/**
* Retrieves entries for the given key range in lexicographic order. Entries will be filtered out by upper bound of given revision
@@ -256,8 +238,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo, long revUpperBound);
+ Cursor<Entry> range(ByteArray keyFrom, @Nullable ByteArray keyTo, long revUpperBound);
/**
* Retrieves entries for the given key range in lexicographic order. Entries will be filtered out by upper bound of given revision
@@ -273,8 +254,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo, long revUpperBound, boolean includeTombstones);
+ Cursor<Entry> range(ByteArray keyFrom, @Nullable ByteArray keyTo, long revUpperBound, boolean includeTombstones);
/**
* Retrieves entries for the given key range in lexicographic order. Short cut for {@link #range(ByteArray, ByteArray, long)} where
@@ -288,8 +268,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo);
+ Cursor<Entry> range(ByteArray keyFrom, @Nullable ByteArray keyTo);
/**
* Retrieves entries for the given key range in lexicographic order. Short cut for
@@ -304,8 +283,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo, boolean includeTombstones);
+ Cursor<Entry> range(ByteArray keyFrom, @Nullable ByteArray keyTo, boolean includeTombstones);
/**
* Subscribes on meta storage updates matching the parameters.
@@ -322,9 +300,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<IgniteUuid> watch(@Nullable ByteArray keyFrom, @Nullable ByteArray keyTo,
- long revision, @NotNull WatchListener lsnr);
+ CompletableFuture<IgniteUuid> watch(@Nullable ByteArray keyFrom, @Nullable ByteArray keyTo, long revision, WatchListener lsnr);
/**
* Subscribes on meta storage updates for the given key.
@@ -340,8 +316,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<IgniteUuid> watch(@NotNull ByteArray key, long revision, @NotNull WatchListener lsnr);
+ CompletableFuture<IgniteUuid> watch(ByteArray key, long revision, WatchListener lsnr);
/**
* Subscribes on meta storage updates for given keys.
@@ -357,8 +332,7 @@ public interface MetaStorageService {
* @see ByteArray
* @see Entry
*/
- @NotNull
- CompletableFuture<IgniteUuid> watch(@NotNull Set<ByteArray> keys, long revision, @NotNull WatchListener lsnr);
+ CompletableFuture<IgniteUuid> watch(Set<ByteArray> keys, long revision, WatchListener lsnr);
/**
* Cancels subscription for the given identifier.
@@ -367,8 +341,7 @@ public interface MetaStorageService {
* @return Completed future in case of operation success. Couldn't be {@code null}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
*/
- @NotNull
- CompletableFuture<Void> stopWatch(@NotNull IgniteUuid id);
+ CompletableFuture<Void> stopWatch(IgniteUuid id);
/**
* Compacts meta storage (removes all tombstone entries and old entries except of entries with latest revision).
@@ -376,7 +349,6 @@ public interface MetaStorageService {
* @return Completed future. Couldn't be {@code null}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
*/
- @NotNull
CompletableFuture<Void> compact();
/**
@@ -386,6 +358,5 @@ public interface MetaStorageService {
* @return Completed future in case of operation success. Couldn't be {@code null}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
*/
- @NotNull
- CompletableFuture<Void> closeCursors(@NotNull String nodeId);
+ CompletableFuture<Void> closeCursors(String nodeId);
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageServiceImpl.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageServiceImpl.java
index fde8834217..f3af7b174e 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageServiceImpl.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageServiceImpl.java
@@ -73,6 +73,8 @@ import org.apache.ignite.internal.metastorage.dsl.If;
import org.apache.ignite.internal.metastorage.dsl.Operation;
import org.apache.ignite.internal.metastorage.dsl.OperationType;
import org.apache.ignite.internal.metastorage.dsl.SimpleCondition;
+import org.apache.ignite.internal.metastorage.dsl.SimpleCondition.RevisionCondition;
+import org.apache.ignite.internal.metastorage.dsl.SimpleCondition.ValueCondition;
import org.apache.ignite.internal.metastorage.dsl.Statement;
import org.apache.ignite.internal.metastorage.dsl.StatementResult;
import org.apache.ignite.internal.metastorage.dsl.Update;
@@ -84,7 +86,6 @@ import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.lang.IgniteUuidGenerator;
import org.apache.ignite.lang.NodeStoppingException;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -129,7 +130,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Entry> get(@NotNull ByteArray key) {
+ public CompletableFuture<Entry> get(ByteArray key) {
GetCommand getCommand = commandsFactory.getCommand().key(key.bytes()).build();
return metaStorageRaftGrpSvc.run(getCommand).thenApply(MetaStorageServiceImpl::singleEntryResult);
@@ -137,7 +138,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Entry> get(@NotNull ByteArray key, long revUpperBound) {
+ public CompletableFuture<Entry> get(ByteArray key, long revUpperBound) {
GetCommand getCommand = commandsFactory.getCommand().key(key.bytes()).revision(revUpperBound).build();
return metaStorageRaftGrpSvc.run(getCommand).thenApply(MetaStorageServiceImpl::singleEntryResult);
@@ -145,7 +146,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAll(Set<ByteArray> keys) {
+ public CompletableFuture<Map<ByteArray, Entry>> getAll(Set<ByteArray> keys) {
GetAllCommand getAllCommand = getAllCommand(commandsFactory, keys, 0);
return metaStorageRaftGrpSvc.run(getAllCommand).thenApply(MetaStorageServiceImpl::multipleEntryResult);
@@ -169,7 +170,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Entry> getAndPut(ByteArray key, byte[] value) {
+ public CompletableFuture<Entry> getAndPut(ByteArray key, byte[] value) {
GetAndPutCommand getAndPutCommand = commandsFactory.getAndPutCommand().key(key.bytes()).value(value).build();
return metaStorageRaftGrpSvc.run(getAndPutCommand).thenApply(MetaStorageServiceImpl::singleEntryResult);
@@ -177,7 +178,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Void> putAll(@NotNull Map<ByteArray, byte[]> vals) {
+ public CompletableFuture<Void> putAll(Map<ByteArray, byte[]> vals) {
PutAllCommand putAllCommand = putAllCommand(commandsFactory, vals);
return metaStorageRaftGrpSvc.run(putAllCommand);
@@ -185,7 +186,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAndPutAll(@NotNull Map<ByteArray, byte[]> vals) {
+ public CompletableFuture<Map<ByteArray, Entry>> getAndPutAll(Map<ByteArray, byte[]> vals) {
GetAndPutAllCommand getAndPutAllCommand = getAndPutAllCommand(commandsFactory, vals);
return metaStorageRaftGrpSvc.run(getAndPutAllCommand).thenApply(MetaStorageServiceImpl::multipleEntryResult);
@@ -193,7 +194,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Void> remove(@NotNull ByteArray key) {
+ public CompletableFuture<Void> remove(ByteArray key) {
RemoveCommand removeCommand = commandsFactory.removeCommand().key(key.bytes()).build();
return metaStorageRaftGrpSvc.run(removeCommand);
@@ -201,7 +202,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Entry> getAndRemove(@NotNull ByteArray key) {
+ public CompletableFuture<Entry> getAndRemove(ByteArray key) {
GetAndRemoveCommand getAndRemoveCommand = commandsFactory.getAndRemoveCommand().key(key.bytes()).build();
return metaStorageRaftGrpSvc.run(getAndRemoveCommand).thenApply(MetaStorageServiceImpl::singleEntryResult);
@@ -209,7 +210,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Void> removeAll(@NotNull Set<ByteArray> keys) {
+ public CompletableFuture<Void> removeAll(Set<ByteArray> keys) {
RemoveAllCommand removeAllCommand = removeAllCommand(commandsFactory, keys);
return metaStorageRaftGrpSvc.run(removeAllCommand);
@@ -217,27 +218,27 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAndRemoveAll(@NotNull Set<ByteArray> keys) {
+ public CompletableFuture<Map<ByteArray, Entry>> getAndRemoveAll(Set<ByteArray> keys) {
GetAndRemoveAllCommand getAndRemoveAllCommand = getAndRemoveAllCommand(commandsFactory, keys);
return metaStorageRaftGrpSvc.run(getAndRemoveAllCommand).thenApply(MetaStorageServiceImpl::multipleEntryResult);
}
@Override
- public @NotNull CompletableFuture<Boolean> invoke(
- @NotNull Condition condition,
- @NotNull Operation success,
- @NotNull Operation failure
+ public CompletableFuture<Boolean> invoke(
+ Condition condition,
+ Operation success,
+ Operation failure
) {
return invoke(condition, List.of(success), List.of(failure));
}
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Boolean> invoke(
- @NotNull Condition condition,
- @NotNull Collection<Operation> success,
- @NotNull Collection<Operation> failure
+ public CompletableFuture<Boolean> invoke(
+ Condition condition,
+ Collection<Operation> success,
+ Collection<Operation> failure
) {
ConditionInfo cond = toConditionInfo(condition, commandsFactory);
@@ -252,7 +253,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<StatementResult> invoke(@NotNull If iif) {
+ public CompletableFuture<StatementResult> invoke(If iif) {
MultiInvokeCommand multiInvokeCommand = commandsFactory.multiInvokeCommand().iif(toIfInfo(iif, commandsFactory)).build();
return metaStorageRaftGrpSvc.run(multiInvokeCommand)
@@ -261,14 +262,14 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo, long revUpperBound) {
+ public Cursor<Entry> range(ByteArray keyFrom, @Nullable ByteArray keyTo, long revUpperBound) {
return range(keyFrom, keyTo, revUpperBound, false);
}
/** {@inheritDoc} */
@Override
- public @NotNull Cursor<Entry> range(
- @NotNull ByteArray keyFrom,
+ public Cursor<Entry> range(
+ ByteArray keyFrom,
@Nullable ByteArray keyTo,
long revUpperBound,
boolean includeTombstones
@@ -293,13 +294,13 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo) {
+ public Cursor<Entry> range(ByteArray keyFrom, @Nullable ByteArray keyTo) {
return range(keyFrom, keyTo, false);
}
/** {@inheritDoc} */
@Override
- public @NotNull Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo, boolean includeTombstones) {
+ public Cursor<Entry> range(ByteArray keyFrom, @Nullable ByteArray keyTo, boolean includeTombstones) {
return new CursorImpl<>(
commandsFactory,
metaStorageRaftGrpSvc,
@@ -320,11 +321,11 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<IgniteUuid> watch(
+ public CompletableFuture<IgniteUuid> watch(
@Nullable ByteArray keyFrom,
@Nullable ByteArray keyTo,
long revision,
- @NotNull WatchListener lsnr
+ WatchListener lsnr
) {
CompletableFuture<IgniteUuid> watchRes = metaStorageRaftGrpSvc.run(commandsFactory.watchRangeKeysCommand()
.keyFrom(keyFrom == null ? null : keyFrom.bytes())
@@ -348,20 +349,20 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<IgniteUuid> watch(
- @NotNull ByteArray key,
+ public CompletableFuture<IgniteUuid> watch(
+ ByteArray key,
long revision,
- @NotNull WatchListener lsnr
+ WatchListener lsnr
) {
return watch(key, null, revision, lsnr);
}
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<IgniteUuid> watch(
- @NotNull Set<ByteArray> keys,
+ public CompletableFuture<IgniteUuid> watch(
+ Set<ByteArray> keys,
long revision,
- @NotNull WatchListener lsnr
+ WatchListener lsnr
) {
CompletableFuture<IgniteUuid> watchRes =
metaStorageRaftGrpSvc.run(watchExactKeysCommand(commandsFactory, keys, revision, localNodeId, uuidGenerator.randomUuid()));
@@ -379,7 +380,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Void> stopWatch(@NotNull IgniteUuid id) {
+ public CompletableFuture<Void> stopWatch(IgniteUuid id) {
return CompletableFuture.runAsync(() -> watchProcessor.stopWatch(id));
}
@@ -387,13 +388,13 @@ public class MetaStorageServiceImpl implements MetaStorageService {
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Void> compact() {
+ public CompletableFuture<Void> compact() {
return null;
}
/** {@inheritDoc} */
@Override
- public @NotNull CompletableFuture<Void> closeCursors(@NotNull String nodeId) {
+ public CompletableFuture<Void> closeCursors(String nodeId) {
return metaStorageRaftGrpSvc.run(commandsFactory.cursorsCloseCommand().nodeId(nodeId).build());
}
@@ -410,14 +411,12 @@ public class MetaStorageServiceImpl implements MetaStorageService {
break;
case REMOVE:
- info.key(op.inner().key()).operationType(OperationType.REMOVE.ordinal());
+ info.key(op.key()).operationType(OperationType.REMOVE.ordinal());
break;
case PUT:
- Operation.PutOp inner = (Operation.PutOp) op.inner();
-
- info.key(inner.key()).value(inner.value()).operationType(OperationType.PUT.ordinal());
+ info.key(op.key()).value(op.value()).operationType(OperationType.PUT.ordinal());
break;
@@ -461,30 +460,18 @@ public class MetaStorageServiceImpl implements MetaStorageService {
.build();
}
- private static ConditionInfo toConditionInfo(@NotNull Condition condition, MetaStorageCommandsFactory commandsFactory) {
+ private static ConditionInfo toConditionInfo(Condition condition, MetaStorageCommandsFactory commandsFactory) {
if (condition instanceof SimpleCondition) {
- SimpleConditionInfoBuilder cnd = commandsFactory.simpleConditionInfo();
-
- Object obj = ((SimpleCondition) condition).inner();
-
- if (obj instanceof SimpleCondition.ExistenceCondition) {
- SimpleCondition.ExistenceCondition inner = (SimpleCondition.ExistenceCondition) obj;
-
- cnd.key(inner.key()).conditionType(inner.type().ordinal());
- } else if (obj instanceof SimpleCondition.TombstoneCondition) {
- SimpleCondition.TombstoneCondition inner = (SimpleCondition.TombstoneCondition) obj;
+ var simpleCondition = (SimpleCondition) condition;
- cnd.key(inner.key()).conditionType(inner.type().ordinal());
- } else if (obj instanceof SimpleCondition.RevisionCondition) {
- SimpleCondition.RevisionCondition inner = (SimpleCondition.RevisionCondition) obj;
+ SimpleConditionInfoBuilder cnd = commandsFactory.simpleConditionInfo()
+ .key(simpleCondition.key())
+ .conditionType(simpleCondition.type().ordinal());
- cnd.key(inner.key()).conditionType(inner.type().ordinal()).revision(inner.revision());
- } else if (obj instanceof SimpleCondition.ValueCondition) {
- SimpleCondition.ValueCondition inner = (SimpleCondition.ValueCondition) obj;
-
- cnd.key(inner.key()).conditionType(inner.type().ordinal()).value(inner.value());
- } else {
- assert false : "Unknown condition type: " + obj.getClass().getSimpleName();
+ if (simpleCondition instanceof SimpleCondition.RevisionCondition) {
+ cnd.revision(((RevisionCondition) simpleCondition).revision());
+ } else if (simpleCondition instanceof SimpleCondition.ValueCondition) {
+ cnd.value(((ValueCondition) simpleCondition).value());
}
return cnd.build();
@@ -509,9 +496,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
Map<ByteArray, Entry> res = new HashMap<>();
for (SingleEntryResponse e : resp.entries()) {
- ByteArray key = new ByteArray(e.key());
-
- res.put(key, new EntryImpl(key, e.value(), e.revision(), e.updateCounter()));
+ res.put(new ByteArray(e.key()), new EntryImpl(e.key(), e.value(), e.revision(), e.updateCounter()));
}
return res;
@@ -528,7 +513,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
private static Entry singleEntryResult(Object obj) {
SingleEntryResponse resp = (SingleEntryResponse) obj;
- return new EntryImpl(new ByteArray(resp.key()), resp.value(), resp.revision(), resp.updateCounter());
+ return new EntryImpl(resp.key(), resp.value(), resp.revision(), resp.updateCounter());
}
private static WatchEvent watchResponse(Object obj) {
@@ -542,7 +527,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
for (int i = 0; i < resp.entries().size(); i++) {
SingleEntryResponse s = resp.entries().get(i);
- EntryImpl e = new EntryImpl(new ByteArray(s.key()), s.value(), s.revision(), s.updateCounter());
+ EntryImpl e = new EntryImpl(s.key(), s.value(), s.revision(), s.updateCounter());
if (i % 2 == 0) {
o = e;
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractCompoundCondition.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractCompoundCondition.java
index 9c4ac40357..df4ead4a2e 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractCompoundCondition.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractCompoundCondition.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.metastorage.server;
import java.util.Arrays;
+import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.util.ArrayUtils;
import org.jetbrains.annotations.NotNull;
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractSimpleCondition.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractSimpleCondition.java
index 772fa8e992..8e8bb55551 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractSimpleCondition.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractSimpleCondition.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.metastorage.server;
+import org.apache.ignite.internal.metastorage.Entry;
import org.jetbrains.annotations.NotNull;
/**
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Condition.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Condition.java
index ec94c40f9c..36066d9a37 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Condition.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Condition.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.metastorage.server;
import java.util.Collection;
-import org.jetbrains.annotations.NotNull;
+import org.apache.ignite.internal.metastorage.Entry;
/**
* Defines interface for boolean condition which could be applied to an array of entries.
@@ -31,7 +31,7 @@ public interface Condition {
*
* @return The keys which identifies an entries which condition will be applied to.
*/
- @NotNull byte[][] keys();
+ byte[][] keys();
/**
* Tests the given entries on condition.
@@ -39,5 +39,5 @@ public interface Condition {
* @param entries Array of entries which will be tested on the condition. Can't be {@code null}.
* @return {@code True} if the given entries satisfies to the condition, otherwise - {@code false}.
*/
- boolean test(@NotNull Entry... entries);
+ boolean test(Entry... entries);
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Entry.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Entry.java
deleted file mode 100644
index c5f20410e8..0000000000
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Entry.java
+++ /dev/null
@@ -1,192 +0,0 @@
-/*
- * 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.metastorage.server;
-
-import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Represents a storage unit as entry with key, value and revision.
- *
- * <p>Where:
- * <ul>
- * <li>key - an unique entry's key represented by an array of bytes. Keys are comparable in lexicographic manner.</li>
- * <li>value - a data which is associated with a key and represented as an array of bytes.</li>
- * <li>revision - a number which denotes a version of whole meta storage.
- * Each change (which could include multiple entries) increments the revision. </li>
- * <li>updateCounter - a number which increments on every update in the change under one revision.</li>
- * </ul>
- *
- * <p>Instance of {@link #Entry} could represents:
- * <ul>
- * <li>A regular entry which stores a particular key, a value and a revision number.</li>
- * <li>An empty entry which denotes absence a regular entry in the meta storage for a given key.
- * A revision is 0 for such kind of entry.</li>
- * <li>A tombstone entry which denotes that a regular entry for a given key was removed from storage on some revision.</li>
- * </ul>
- */
-public class Entry {
- /** Entry key. Couldn't be {@code null}. */
- @NotNull
- private final byte[] key;
-
- /**
- * Entry value.
- * <p>
- * {@code val == null} only for {@link #empty()} and {@link #tombstone()} entries.
- * </p>
- */
- @Nullable
- private final byte[] val;
-
- /**
- * Revision number corresponding to this particular entry.
- * <p>
- * {@code rev == 0} for {@link #empty()} entry, {@code rev > 0} for regular and {@link #tombstone()} entries.
- * </p>
- */
- private final long rev;
-
- /**
- * Update counter corresponds to this particular entry.
- * <p>
- * {@code updCntr == 0} for {@link #empty()} entry, {@code updCntr > 0} for regular and {@link #tombstone()} entries.
- * </p>
- */
- private final long updCntr;
-
- /**
- * Constructor.
- *
- * @param key Key bytes. Couldn't be {@code null}.
- * @param val Value bytes. Couldn't be {@code null}.
- * @param rev Revision.
- * @param updCntr Update counter.
- */
- // TODO: It seems user will never create Entry, so we can reduce constructor scope to protected or package-private
- // and reuse it from two-place private constructor.
- public Entry(@NotNull byte[] key, @NotNull byte[] val, long rev, long updCntr) {
- assert key != null : "key can't be null";
- assert val != null : "value can't be null";
-
- this.key = key;
- this.val = val;
- this.rev = rev;
- this.updCntr = updCntr;
- }
-
- /**
- * Constructor for empty and tombstone entries.
- *
- * @param key Key bytes. Couldn't be {@code null}.
- * @param rev Revision.
- * @param updCntr Update counter.
- */
- private Entry(@NotNull byte[] key, long rev, long updCntr) {
- assert key != null : "key can't be null";
-
- this.key = key;
- this.val = null;
- this.rev = rev;
- this.updCntr = updCntr;
- }
-
- /**
- * Creates an instance of empty entry for a given key.
- *
- * @param key Key bytes. Couldn't be {@code null}.
- * @return Empty entry.
- */
- @NotNull
- public static Entry empty(byte[] key) {
- return new Entry(key, 0, 0);
- }
-
- /**
- * Returns value which denotes whether entry is empty or not.
- *
- * @return {@code True} if entry is empty, otherwise - {@code false}.
- */
- public boolean empty() {
- return val == null && rev == 0 && updCntr == 0;
- }
-
- /**
- * Creates an instance of tombstone entry for a given key and a revision.
- *
- * @param key Key bytes. Couldn't be {@code null}.
- * @param rev Revision.
- * @param updCntr Update counter.
- * @return Empty entry.
- */
- @NotNull
- public static Entry tombstone(byte[] key, long rev, long updCntr) {
- assert rev > 0 : "rev must be positive for tombstone entry.";
- assert updCntr > 0 : "updCntr must be positive for tombstone entry.";
-
- return new Entry(key, rev, updCntr);
- }
-
- /**
- * Returns value which denotes whether entry is tombstone or not.
- *
- * @return {@code True} if entry is tombstone, otherwise - {@code false}.
- */
- public boolean tombstone() {
- return val == null && rev > 0 && updCntr > 0;
- }
-
-
- /**
- * Returns a key.
- *
- * @return Key.
- */
- @NotNull
- public byte[] key() {
- return key;
- }
-
- /**
- * Returns a value.
- *
- * @return Value.
- */
- @Nullable
- public byte[] value() {
- return val;
- }
-
- /**
- * Returns a revision.
- *
- * @return Revision.
- */
- public long revision() {
- return rev;
- }
-
- /**
- * Returns a update counter.
- *
- * @return Update counter.
- */
- public long updateCounter() {
- return updCntr;
- }
-}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java
deleted file mode 100644
index 6a66dd8249..0000000000
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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.metastorage.server;
-
-/**
- * Represent an update event for particular key and entry.
- */
-public class EntryEvent {
- /** Old (previous) entry. */
- private final Entry oldEntry;
-
- /** New (current) entry. */
- private final Entry entry;
-
- /**
- * Constructs event with given old and new entries.
- *
- * @param oldEntry Old entry.
- * @param curEntry New entry.
- */
- public EntryEvent(Entry oldEntry, Entry curEntry) {
- this.oldEntry = oldEntry;
- this.entry = curEntry;
- }
-
- /**
- * Returns old entry.
- *
- * @return Old entry.
- */
- public Entry oldEntry() {
- return oldEntry;
- }
-
- /**
- * Rreturns new entry.
- *
- * @return New entry.
- */
- public Entry entry() {
- return entry;
- }
-}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/ExistenceCondition.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/ExistenceCondition.java
index 4827aadf2e..bf338c01f3 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/ExistenceCondition.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/ExistenceCondition.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.metastorage.server;
+import org.apache.ignite.internal.metastorage.Entry;
import org.jetbrains.annotations.NotNull;
/**
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/If.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/If.java
index f0f9c51870..62014f5da0 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/If.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/If.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.metastorage.server;
+import org.apache.ignite.internal.metastorage.dsl.Update;
+
/**
* Root building block for the compound meta storage invoke command.
* Contains of boolean condition and 2 branches of execution, like usual programming language's if.
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java
index 41154a601c..0b281095cd 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java
@@ -22,8 +22,11 @@ import java.util.Collection;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.close.ManuallyCloseable;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.WatchEvent;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.metastorage.dsl.StatementResult;
import org.apache.ignite.internal.util.Cursor;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -55,7 +58,7 @@ public interface KeyValueStorage extends ManuallyCloseable {
* @param key The key.
* @return Value corresponding to the given key.
*/
- @NotNull Entry get(byte[] key);
+ Entry get(byte[] key);
/**
* Returns an entry by the given key and bounded by the given revision.
@@ -64,7 +67,7 @@ public interface KeyValueStorage extends ManuallyCloseable {
* @param revUpperBound The upper bound of revision.
* @return Value corresponding to the given key.
*/
- @NotNull Entry get(byte[] key, long revUpperBound);
+ Entry get(byte[] key, long revUpperBound);
/**
* Returns all entries corresponding to given keys.
@@ -72,7 +75,7 @@ public interface KeyValueStorage extends ManuallyCloseable {
* @param keys Keys collection.
* @return Entries corresponding to given keys.
*/
- @NotNull Collection<Entry> getAll(List<byte[]> keys);
+ Collection<Entry> getAll(List<byte[]> keys);
/**
* Returns all entries corresponding to given keys and bounded by the given revision.
@@ -81,7 +84,7 @@ public interface KeyValueStorage extends ManuallyCloseable {
* @param revUpperBound Upper bound of revision.
* @return Entries corresponding to given keys.
*/
- @NotNull Collection<Entry> getAll(List<byte[]> keys, long revUpperBound);
+ Collection<Entry> getAll(List<byte[]> keys, long revUpperBound);
/**
* Inserts an entry with the given key and given value.
@@ -98,7 +101,7 @@ public interface KeyValueStorage extends ManuallyCloseable {
* @param value The value.
* @return Previous entry corresponding to the given key.
*/
- @NotNull Entry getAndPut(byte[] key, byte[] value);
+ Entry getAndPut(byte[] key, byte[] value);
/**
* Inserts entries with given keys and given values.
@@ -115,7 +118,7 @@ public interface KeyValueStorage extends ManuallyCloseable {
* @param values The values list.
* @return Collection of previous entries corresponding to given keys.
*/
- @NotNull Collection<Entry> getAndPutAll(List<byte[]> keys, List<byte[]> values);
+ Collection<Entry> getAndPutAll(List<byte[]> keys, List<byte[]> values);
/**
* Removes an entry with the given key.
@@ -130,7 +133,7 @@ public interface KeyValueStorage extends ManuallyCloseable {
* @param key The key.
* @return Previous entry.
*/
- @NotNull Entry getAndRemove(byte[] key);
+ Entry getAndRemove(byte[] key);
/**
* Remove all entries corresponding to given keys.
@@ -145,7 +148,7 @@ public interface KeyValueStorage extends ManuallyCloseable {
* @param keys The keys list.
* @return Previous entries.
*/
- @NotNull Collection<Entry> getAndRemoveAll(List<byte[]> keys);
+ Collection<Entry> getAndRemoveAll(List<byte[]> keys);
/**
* Performs {@code success} operation if condition is {@code true}, otherwise performs {@code failure} operations.
@@ -166,7 +169,7 @@ public interface KeyValueStorage extends ManuallyCloseable {
* @see If
* @see StatementResult
*/
- @NotNull StatementResult invoke(@NotNull If iif);
+ StatementResult invoke(If iif);
/**
* Returns cursor by entries which correspond to the given keys range.
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Operation.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Operation.java
deleted file mode 100644
index b0c1976981..0000000000
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Operation.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * 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.metastorage.server;
-
-import org.apache.ignite.internal.metastorage.dsl.OperationType;
-import org.apache.ignite.internal.util.IgniteUtils;
-import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Defines operation which will be applied to an entry identified by the key.
- *
- * <p>Invariants:
- * <ul>
- * <li>Any operation identifies a target entry by not null {@code key} except of {@link OperationType#NO_OP}.</li>
- * <li>Only {@link OperationType#PUT} operation contains value which will be written to meta storage.</li>
- * </ul>
- */
-public final class Operation {
- /**
- * Key identifies an entry which operation will be applied to. Key is {@code null} for {@link OperationType#NO_OP} operation.
- */
- @Nullable
- private final byte[] key;
-
- /**
- * Value which will be associated with the {@link #key}. Value is not {@code null} only for {@link OperationType#PUT} operation.
- */
- @Nullable
- private final byte[] val;
-
- /**
- * Operation type.
- *
- * @see OperationType
- */
- @NotNull
- private final OperationType type;
-
- /**
- * Constructs operation which will be applied to an entry identified by the given key.
- *
- * @param type Operation type. Can't be {@code null}.
- * @param key Key identifies an entry which operation will be applied to.
- * @param val Value will be associated with an entry identified by the {@code key}.
- */
- public Operation(@NotNull OperationType type, byte @Nullable [] key, byte @Nullable [] val) {
- assert (type == OperationType.NO_OP && key == null && val == null)
- || (type == OperationType.PUT && key != null && val != null)
- || (type == OperationType.REMOVE && key != null && val == null)
- : "Invalid operation parameters: [type=" + type
- + ", key=" + (key == null ? "null" : IgniteUtils.toHexString(key, 256))
- + ", val=" + (val == null ? "null" : IgniteUtils.toHexString(val, 256)) + ']';
-
- this.key = key;
- this.val = val;
- this.type = type;
- }
-
- /**
- * Returns a key which identifies an entry which operation will be applied to.
- *
- * @return A key which identifies an entry which operation will be applied to.
- */
- @Nullable
- public byte[] key() {
- return key;
- }
-
- /**
- * Returns a value which will be associated with an entry identified by the {@code key}.
- *
- * @return A value which will be associated with an entry identified by the {@code key}.
- */
- @Nullable
- public byte[] value() {
- return val;
- }
-
- /**
- * Returns an operation type.
- *
- * @return An operation type.
- */
- @NotNull
- public OperationType type() {
- return type;
- }
-}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/RevisionCondition.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/RevisionCondition.java
index 76a284889e..746133d273 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/RevisionCondition.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/RevisionCondition.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.metastorage.server;
+import org.apache.ignite.internal.metastorage.Entry;
import org.jetbrains.annotations.NotNull;
/**
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Statement.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Statement.java
index e0affab47e..1cc20587f2 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Statement.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Statement.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.metastorage.server;
+import org.apache.ignite.internal.metastorage.dsl.Update;
+
/**
* Simple Either-like wrapper to hold one of the statement type: {@link If} or {@link Update}.
* Needed to construct and simple deconstruction of nested {@link If},
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/StatementResult.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/StatementResult.java
deleted file mode 100644
index e227a02c95..0000000000
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/StatementResult.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.metastorage.server;
-
-import java.nio.ByteBuffer;
-
-/**
- * Simple result of statement execution, backed by byte[] array.
- * Provides some shortcut methods to represent the values of some primitive types.
- */
-public class StatementResult {
- /** Result data. */
- private final byte[] res;
-
- /**
- * Constructs result from the byte array.
- *
- * @param res byte array.
- */
- public StatementResult(byte[] res) {
- this.res = res;
- }
-
- /**
- * Constructs result from the boolean value.
- *
- * @param res boolean.
- */
- public StatementResult(boolean res) {
- this.res = new byte[] {(byte) (res ? 1 : 0)};
- }
-
- /**
- * Constructs result from the int value.
- *
- * @param res int.
- */
- public StatementResult(int res) {
- this.res = ByteBuffer.allocate(4).putInt(res).array();
- }
-
- /**
- * Returns backed byte array.
- *
- * @return backed byte array.
- */
- public byte[] bytes() {
- return res;
- }
-}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/TombstoneCondition.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/TombstoneCondition.java
index 8d60e38061..cb0f9d7fd5 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/TombstoneCondition.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/TombstoneCondition.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.metastorage.server;
+import org.apache.ignite.internal.metastorage.Entry;
import org.jetbrains.annotations.NotNull;
/**
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Update.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Update.java
deleted file mode 100644
index 4801f41c50..0000000000
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Update.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.metastorage.server;
-
-import java.util.Collection;
-
-/**
- * Simple operations + result wrapper to describe the terminal branch of {@link If} execution.
- */
-public class Update {
- /** Operations. */
- private final Collection<Operation> ops;
-
- /** Result. */
- private final StatementResult result;
-
- /**
- * Constructs new update object.
- *
- * @param ops operations
- * @param result result
- */
- public Update(Collection<Operation> ops, StatementResult result) {
- this.ops = ops;
- this.result = result;
- }
-
- /**
- * Returns operations.
- *
- * @return operations.
- */
- public Collection<Operation> operations() {
- return ops;
- }
-
- /**
- * Returns result.
- *
- * @return result.
- */
- public StatementResult result() {
- return result;
- }
-}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/ValueCondition.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/ValueCondition.java
index 49a009d3a4..43a512435d 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/ValueCondition.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/ValueCondition.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.metastorage.server;
import java.util.Arrays;
+import org.apache.ignite.internal.metastorage.Entry;
import org.jetbrains.annotations.NotNull;
/**
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchEvent.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchEvent.java
deleted file mode 100644
index 42a9d8a955..0000000000
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchEvent.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * 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.metastorage.server;
-
-import java.util.Collection;
-import java.util.List;
-
-/**
- * Watch event contains all entry updates done under one revision. Each particular entry update in this revision is represented by {@link
- * EntryEvent} entity.
- */
-public class WatchEvent {
- /** Events about each entry update in the revision. */
- private final List<EntryEvent> entryEvts;
-
- /** Designates that watch event contains only one update revision. */
- private final boolean single;
-
- /**
- * Constructs an watch event with given entry events collection.
- *
- * @param entryEvts Events for entries corresponding to an update under one revision.
- */
- public WatchEvent(List<EntryEvent> entryEvts) {
- assert entryEvts != null && !entryEvts.isEmpty();
-
- this.single = entryEvts.size() == 1;
- this.entryEvts = entryEvts;
- }
-
- /**
- * Returns {@code true} if watch event contains only one entry event.
- *
- * @return {@code True} if watch event contains only one entry event.
- */
- public boolean single() {
- return single;
- }
-
- /**
- * Returns collection of entry entry event done under one revision.
- *
- * @return Collection of entry entry event done under one revision.
- */
- public Collection<EntryEvent> entryEvents() {
- return entryEvts;
- }
-
- /**
- * Returns entry event. It is useful method in case when we know that only one event was modified.
- *
- * @return Entry event.
- */
- public EntryEvent entryEvent() {
- return entryEvts.get(0);
- }
-}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RangeCursor.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RangeCursor.java
index 941103b84d..a0dfc6b185 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RangeCursor.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RangeCursor.java
@@ -21,7 +21,7 @@ import java.util.Arrays;
import java.util.Iterator;
import java.util.Map;
import java.util.NoSuchElementException;
-import org.apache.ignite.internal.metastorage.server.Entry;
+import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.util.Cursor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java
index 411f958807..6f9861fd08 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java
@@ -49,17 +49,18 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.WatchEvent;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.metastorage.dsl.StatementResult;
+import org.apache.ignite.internal.metastorage.dsl.Update;
import org.apache.ignite.internal.metastorage.exceptions.MetaStorageException;
+import org.apache.ignite.internal.metastorage.impl.EntryImpl;
import org.apache.ignite.internal.metastorage.server.Condition;
-import org.apache.ignite.internal.metastorage.server.Entry;
import org.apache.ignite.internal.metastorage.server.If;
import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
-import org.apache.ignite.internal.metastorage.server.Operation;
import org.apache.ignite.internal.metastorage.server.Statement;
-import org.apache.ignite.internal.metastorage.server.StatementResult;
-import org.apache.ignite.internal.metastorage.server.Update;
import org.apache.ignite.internal.metastorage.server.Value;
-import org.apache.ignite.internal.metastorage.server.WatchEvent;
import org.apache.ignite.internal.rocksdb.ColumnFamily;
import org.apache.ignite.internal.rocksdb.RocksBiPredicate;
import org.apache.ignite.internal.rocksdb.RocksUtils;
@@ -849,7 +850,7 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
}
if (revs == null || revs.length == 0) {
- return Entry.empty(key);
+ return EntryImpl.empty(key);
}
long lastRev;
@@ -862,7 +863,7 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
// lastRev can be -1 if maxRevision return -1.
if (lastRev == -1) {
- return Entry.empty(key);
+ return EntryImpl.empty(key);
}
return doGetValue(key, lastRev);
@@ -915,7 +916,7 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
@NotNull
Entry doGetValue(byte[] key, long revision) {
if (revision == 0) {
- return Entry.empty(key);
+ return EntryImpl.empty(key);
}
byte[] valueBytes;
@@ -927,16 +928,16 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
}
if (valueBytes == null || valueBytes.length == 0) {
- return Entry.empty(key);
+ return EntryImpl.empty(key);
}
Value lastVal = bytesToValue(valueBytes);
if (lastVal.tombstone()) {
- return Entry.tombstone(key, revision, lastVal.updateCounter());
+ return EntryImpl.tombstone(key, revision, lastVal.updateCounter());
}
- return new Entry(key, lastVal.bytes(), revision, lastVal.updateCounter());
+ return new EntryImpl(key, lastVal.bytes(), revision, lastVal.updateCounter());
}
/**
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/WatchCursor.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/WatchCursor.java
index fd1b6dd3dc..c63f628715 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/WatchCursor.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/WatchCursor.java
@@ -27,11 +27,12 @@ import java.util.ArrayList;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.function.Predicate;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.EntryEvent;
+import org.apache.ignite.internal.metastorage.WatchEvent;
import org.apache.ignite.internal.metastorage.exceptions.MetaStorageException;
-import org.apache.ignite.internal.metastorage.server.Entry;
-import org.apache.ignite.internal.metastorage.server.EntryEvent;
+import org.apache.ignite.internal.metastorage.impl.EntryImpl;
import org.apache.ignite.internal.metastorage.server.Value;
-import org.apache.ignite.internal.metastorage.server.WatchEvent;
import org.apache.ignite.internal.rocksdb.RocksUtils;
import org.apache.ignite.internal.util.Cursor;
import org.rocksdb.ReadOptions;
@@ -152,9 +153,9 @@ class WatchCursor implements Cursor<WatchEvent> {
Entry newEntry;
if (val.tombstone()) {
- newEntry = Entry.tombstone(key, revision, val.updateCounter());
+ newEntry = EntryImpl.tombstone(key, revision, val.updateCounter());
} else {
- newEntry = new Entry(key, val.bytes(), revision, val.updateCounter());
+ newEntry = new EntryImpl(key, val.bytes(), revision, val.updateCounter());
}
Entry oldEntry = storage.doGet(key, revision - 1);
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/raft/MetaStorageListener.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/raft/MetaStorageListener.java
index 9e9c5cf4a0..44362068c5 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/raft/MetaStorageListener.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/raft/MetaStorageListener.java
@@ -30,6 +30,9 @@ import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.EntryEvent;
+import org.apache.ignite.internal.metastorage.WatchEvent;
import org.apache.ignite.internal.metastorage.command.GetAllCommand;
import org.apache.ignite.internal.metastorage.command.GetAndPutAllCommand;
import org.apache.ignite.internal.metastorage.command.GetAndPutCommand;
@@ -61,23 +64,20 @@ import org.apache.ignite.internal.metastorage.command.info.StatementInfo;
import org.apache.ignite.internal.metastorage.command.info.UpdateInfo;
import org.apache.ignite.internal.metastorage.dsl.CompoundConditionType;
import org.apache.ignite.internal.metastorage.dsl.ConditionType;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.metastorage.dsl.StatementResult;
+import org.apache.ignite.internal.metastorage.dsl.Update;
import org.apache.ignite.internal.metastorage.exceptions.MetaStorageException;
import org.apache.ignite.internal.metastorage.server.AndCondition;
import org.apache.ignite.internal.metastorage.server.Condition;
-import org.apache.ignite.internal.metastorage.server.Entry;
-import org.apache.ignite.internal.metastorage.server.EntryEvent;
import org.apache.ignite.internal.metastorage.server.ExistenceCondition;
import org.apache.ignite.internal.metastorage.server.If;
import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
-import org.apache.ignite.internal.metastorage.server.Operation;
import org.apache.ignite.internal.metastorage.server.OrCondition;
import org.apache.ignite.internal.metastorage.server.RevisionCondition;
import org.apache.ignite.internal.metastorage.server.Statement;
-import org.apache.ignite.internal.metastorage.server.StatementResult;
import org.apache.ignite.internal.metastorage.server.TombstoneCondition;
-import org.apache.ignite.internal.metastorage.server.Update;
import org.apache.ignite.internal.metastorage.server.ValueCondition;
-import org.apache.ignite.internal.metastorage.server.WatchEvent;
import org.apache.ignite.internal.raft.ReadCommand;
import org.apache.ignite.internal.raft.WriteCommand;
import org.apache.ignite.internal.raft.service.CommandClosure;
@@ -304,7 +304,7 @@ public class MetaStorageListener implements RaftGroupListener {
for (EntryEvent e : evt.entryEvents()) {
Entry o = e.oldEntry();
- Entry n = e.entry();
+ Entry n = e.newEntry();
resp.add(new SingleEntryResponse(o.key(), o.value(), o.revision(), o.updateCounter()));
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java
index 5b9b811113..4cbd89c6c7 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java
@@ -192,7 +192,7 @@ public class WatchAggregator {
var filteredEvts = new ArrayList<EntryEvent>();
for (EntryEvent entryEvt : evt.entryEvents()) {
- if (watch.keyCriterion().contains(entryEvt.oldEntry().key())) {
+ if (watch.keyCriterion().contains(new ByteArray(entryEvt.oldEntry().key()))) {
filteredEvts.add(entryEvt);
}
}
@@ -217,7 +217,7 @@ public class WatchAggregator {
for (EntryEvent entryEvt : evt.entryEvents()) {
revision = entryEvt.newEntry().revision();
- entries.add(new IgniteBiTuple<>(entryEvt.newEntry().key(), entryEvt.newEntry().value()));
+ entries.add(new IgniteBiTuple<>(new ByteArray(entryEvt.newEntry().key()), entryEvt.newEntry().value()));
}
storeRevision.accept(entries, revision);
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/impl/MetaStorageRangeCursorTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/impl/MetaStorageRangeCursorTest.java
index a7340c53a0..d948732013 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/impl/MetaStorageRangeCursorTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/impl/MetaStorageRangeCursorTest.java
@@ -39,7 +39,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.IntStream;
import java.util.stream.Stream;
-import org.apache.ignite.internal.metastorage.server.Entry;
+import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageListener;
import org.apache.ignite.internal.raft.Command;
@@ -103,15 +103,15 @@ public class MetaStorageRangeCursorTest {
return args.stream();
}
- private void checkCursor(Cursor<org.apache.ignite.internal.metastorage.Entry> range, int count) {
+ private void checkCursor(Cursor<Entry> range, int count) {
for (int i = 0; i < count; i++) {
String errorDetails = "count=" + count + ", i=" + i;
assertTrue(range.hasNext(), errorDetails);
- org.apache.ignite.internal.metastorage.Entry e = range.next();
+ Entry e = range.next();
- assertEquals(intToBytes(i), e.key(), errorDetails);
+ assertArrayEquals(intToBytes(i).bytes(), e.key(), errorDetails);
assertArrayEquals(intToBytes(i).bytes(), e.value(), errorDetails);
assertEquals(i, e.revision(), errorDetails);
assertEquals(i, e.updateCounter(), errorDetails);
@@ -125,7 +125,7 @@ public class MetaStorageRangeCursorTest {
}
private Entry intToEntry(int i) {
- return new Entry(intToBytes(i).bytes(), intToBytes(i).bytes(), i, i);
+ return new EntryImpl(intToBytes(i).bytes(), intToBytes(i).bytes(), i, i);
}
private ByteArray intToBytes(int i) {
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorageTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorageTest.java
index ce2605701c..e4736c8a5c 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorageTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorageTest.java
@@ -35,7 +35,13 @@ import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.stream.Collectors;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.EntryEvent;
+import org.apache.ignite.internal.metastorage.WatchEvent;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
import org.apache.ignite.internal.metastorage.dsl.OperationType;
+import org.apache.ignite.internal.metastorage.dsl.StatementResult;
+import org.apache.ignite.internal.metastorage.dsl.Update;
import org.apache.ignite.internal.metastorage.server.ValueCondition.Type;
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.lang.ByteArray;
@@ -75,7 +81,7 @@ public abstract class AbstractKeyValueStorageTest {
@Test
public void put() {
byte[] key = key(1);
- final byte[] val = keyValue(1, 1);
+ byte[] val = keyValue(1, 1);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -111,14 +117,14 @@ public abstract class AbstractKeyValueStorageTest {
byte[] key1 = key(1);
byte[] val1 = keyValue(1, 1);
- final byte[] key2 = key(2);
- final byte[] val2_1 = keyValue(2, 21);
- final byte[] val2_2 = keyValue(2, 22);
+ byte[] key2 = key(2);
+ byte[] val21 = keyValue(2, 21);
+ byte[] val22 = keyValue(2, 22);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
- final byte[] key4 = key(4);
+ byte[] key4 = key(4);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -127,8 +133,8 @@ public abstract class AbstractKeyValueStorageTest {
storage.put(key1, val1);
// Rewrite.
- storage.put(key2, val2_1);
- storage.put(key2, val2_2);
+ storage.put(key2, val21);
+ storage.put(key2, val22);
// Remove.
storage.put(key3, val3);
@@ -152,7 +158,7 @@ public abstract class AbstractKeyValueStorageTest {
assertNotNull(key2EntryBounded2);
assertEquals(2, key2EntryBounded2.revision());
assertEquals(2, key2EntryBounded2.updateCounter());
- assertArrayEquals(val2_1, key2EntryBounded2.value());
+ assertArrayEquals(val21, key2EntryBounded2.value());
assertFalse(key2EntryBounded2.tombstone());
assertFalse(key2EntryBounded2.empty());
@@ -180,7 +186,7 @@ public abstract class AbstractKeyValueStorageTest {
assertNotNull(key2EntryBounded5);
assertEquals(3, key2EntryBounded5.revision());
assertEquals(3, key2EntryBounded5.updateCounter());
- assertArrayEquals(val2_2, key2EntryBounded5.value());
+ assertArrayEquals(val22, key2EntryBounded5.value());
assertFalse(key2EntryBounded5.tombstone());
assertFalse(key2EntryBounded5.empty());
@@ -199,14 +205,14 @@ public abstract class AbstractKeyValueStorageTest {
byte[] key1 = key(1);
byte[] val1 = keyValue(1, 1);
- final byte[] key2 = key(2);
- final byte[] val2_1 = keyValue(2, 21);
- final byte[] val2_2 = keyValue(2, 22);
+ byte[] key2 = key(2);
+ byte[] val21 = keyValue(2, 21);
+ byte[] val22 = keyValue(2, 22);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
- final byte[] key4 = key(4);
+ byte[] key4 = key(4);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -215,8 +221,8 @@ public abstract class AbstractKeyValueStorageTest {
storage.put(key1, val1);
// Rewrite.
- storage.put(key2, val2_1);
- storage.put(key2, val2_2);
+ storage.put(key2, val21);
+ storage.put(key2, val22);
// Remove.
storage.put(key3, val3);
@@ -249,7 +255,7 @@ public abstract class AbstractKeyValueStorageTest {
assertEquals(3, e2.updateCounter());
assertFalse(e2.tombstone());
assertFalse(e2.empty());
- assertArrayEquals(val2_2, e2.value());
+ assertArrayEquals(val22, e2.value());
// Test removed value.
Entry e3 = map.get(new ByteArray(key3));
@@ -273,14 +279,14 @@ public abstract class AbstractKeyValueStorageTest {
byte[] key1 = key(1);
byte[] val1 = keyValue(1, 1);
- final byte[] key2 = key(2);
- final byte[] val2_1 = keyValue(2, 21);
- final byte[] val2_2 = keyValue(2, 22);
+ byte[] key2 = key(2);
+ byte[] val21 = keyValue(2, 21);
+ byte[] val22 = keyValue(2, 22);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
- final byte[] key4 = key(4);
+ byte[] key4 = key(4);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -289,8 +295,8 @@ public abstract class AbstractKeyValueStorageTest {
storage.put(key1, val1);
// Rewrite.
- storage.put(key2, val2_1);
- storage.put(key2, val2_2);
+ storage.put(key2, val21);
+ storage.put(key2, val22);
// Remove.
storage.put(key3, val3);
@@ -324,7 +330,7 @@ public abstract class AbstractKeyValueStorageTest {
assertEquals(2, e2.updateCounter());
assertFalse(e2.tombstone());
assertFalse(e2.empty());
- assertArrayEquals(val2_1, e2.value());
+ assertArrayEquals(val21, e2.value());
// Values with larger revision don't exist yet.
Entry e3 = map.get(new ByteArray(key3));
@@ -362,7 +368,7 @@ public abstract class AbstractKeyValueStorageTest {
assertEquals(3, e2.updateCounter());
assertFalse(e2.tombstone());
assertFalse(e2.empty());
- assertArrayEquals(val2_2, e2.value());
+ assertArrayEquals(val22, e2.value());
// Test not removed value.
e3 = map.get(new ByteArray(key3));
@@ -384,7 +390,7 @@ public abstract class AbstractKeyValueStorageTest {
@Test
public void getAndPut() {
byte[] key = key(1);
- final byte[] val = keyValue(1, 1);
+ byte[] val = keyValue(1, 1);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -411,18 +417,18 @@ public abstract class AbstractKeyValueStorageTest {
@Test
public void putAll() {
- final byte[] key1 = key(1);
- final byte[] val1 = keyValue(1, 1);
+ byte[] key1 = key(1);
+ byte[] val1 = keyValue(1, 1);
byte[] key2 = key(2);
byte[] val21 = keyValue(2, 21);
- final byte[] val2_2 = keyValue(2, 22);
+ byte[] val22 = keyValue(2, 22);
- final byte[] key3 = key(3);
- final byte[] val3_1 = keyValue(3, 31);
- final byte[] val3_2 = keyValue(3, 32);
+ byte[] key3 = key(3);
+ byte[] val31 = keyValue(3, 31);
+ byte[] val32 = keyValue(3, 32);
- final byte[] key4 = key(4);
+ byte[] key4 = key(4);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -431,13 +437,13 @@ public abstract class AbstractKeyValueStorageTest {
storage.put(key2, val21);
// Remove. Tombstone must be replaced by new value.
- storage.put(key3, val3_1);
+ storage.put(key3, val31);
storage.remove(key3);
assertEquals(3, storage.revision());
assertEquals(3, storage.updateCounter());
- storage.putAll(List.of(key1, key2, key3), List.of(val1, val2_2, val3_2));
+ storage.putAll(List.of(key1, key2, key3), List.of(val1, val22, val32));
assertEquals(4, storage.revision());
assertEquals(6, storage.updateCounter());
@@ -466,7 +472,7 @@ public abstract class AbstractKeyValueStorageTest {
assertEquals(5, e2.updateCounter());
assertFalse(e2.tombstone());
assertFalse(e2.empty());
- assertArrayEquals(val2_2, e2.value());
+ assertArrayEquals(val22, e2.value());
// Test removed value.
Entry e3 = map.get(new ByteArray(key3));
@@ -487,18 +493,18 @@ public abstract class AbstractKeyValueStorageTest {
@Test
public void getAndPutAll() {
- final byte[] key1 = key(1);
- final byte[] val1 = keyValue(1, 1);
+ byte[] key1 = key(1);
+ byte[] val1 = keyValue(1, 1);
byte[] key2 = key(2);
byte[] val21 = keyValue(2, 21);
- final byte[] val2_2 = keyValue(2, 22);
+ byte[] val22 = keyValue(2, 22);
- final byte[] key3 = key(3);
- final byte[] val3_1 = keyValue(3, 31);
- final byte[] val3_2 = keyValue(3, 32);
+ byte[] key3 = key(3);
+ byte[] val31 = keyValue(3, 31);
+ byte[] val32 = keyValue(3, 32);
- final byte[] key4 = key(4);
+ byte[] key4 = key(4);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -507,13 +513,13 @@ public abstract class AbstractKeyValueStorageTest {
storage.put(key2, val21);
// Remove. Tombstone must be replaced by new value.
- storage.put(key3, val3_1);
+ storage.put(key3, val31);
storage.remove(key3);
assertEquals(3, storage.revision());
assertEquals(3, storage.updateCounter());
- Collection<Entry> entries = storage.getAndPutAll(List.of(key1, key2, key3), List.of(val1, val2_2, val3_2));
+ Collection<Entry> entries = storage.getAndPutAll(List.of(key1, key2, key3), List.of(val1, val22, val32));
assertEquals(4, storage.revision());
assertEquals(6, storage.updateCounter());
@@ -575,7 +581,7 @@ public abstract class AbstractKeyValueStorageTest {
assertEquals(5, e2.updateCounter());
assertFalse(e2.tombstone());
assertFalse(e2.empty());
- assertArrayEquals(val2_2, e2.value());
+ assertArrayEquals(val22, e2.value());
// Test removed value.
e3 = map.get(new ByteArray(key3));
@@ -597,7 +603,7 @@ public abstract class AbstractKeyValueStorageTest {
@Test
public void remove() {
byte[] key = key(1);
- final byte[] val = keyValue(1, 1);
+ byte[] val = keyValue(1, 1);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -645,7 +651,7 @@ public abstract class AbstractKeyValueStorageTest {
@Test
public void getAndRemove() {
byte[] key = key(1);
- final byte[] val = keyValue(1, 1);
+ byte[] val = keyValue(1, 1);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -704,14 +710,14 @@ public abstract class AbstractKeyValueStorageTest {
byte[] key1 = key(1);
byte[] val1 = keyValue(1, 1);
- final byte[] key2 = key(2);
- final byte[] val2_1 = keyValue(2, 21);
- final byte[] val2_2 = keyValue(2, 22);
+ byte[] key2 = key(2);
+ byte[] val21 = keyValue(2, 21);
+ byte[] val22 = keyValue(2, 22);
- final byte[] key3 = key(3);
- final byte[] val3_1 = keyValue(3, 31);
+ byte[] key3 = key(3);
+ byte[] val31 = keyValue(3, 31);
- final byte[] key4 = key(4);
+ byte[] key4 = key(4);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -720,11 +726,11 @@ public abstract class AbstractKeyValueStorageTest {
storage.put(key1, val1);
// Rewrite.
- storage.put(key2, val2_1);
- storage.put(key2, val2_2);
+ storage.put(key2, val21);
+ storage.put(key2, val22);
// Remove. Tombstone must not be removed again.
- storage.put(key3, val3_1);
+ storage.put(key3, val31);
storage.remove(key3);
assertEquals(5, storage.revision());
@@ -781,14 +787,14 @@ public abstract class AbstractKeyValueStorageTest {
byte[] key1 = key(1);
byte[] val1 = keyValue(1, 1);
- final byte[] key2 = key(2);
- final byte[] val2_1 = keyValue(2, 21);
- final byte[] val2_2 = keyValue(2, 22);
+ byte[] key2 = key(2);
+ byte[] val21 = keyValue(2, 21);
+ byte[] val22 = keyValue(2, 22);
- final byte[] key3 = key(3);
- final byte[] val3_1 = keyValue(3, 31);
+ byte[] key3 = key(3);
+ byte[] val31 = keyValue(3, 31);
- final byte[] key4 = key(4);
+ byte[] key4 = key(4);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -797,11 +803,11 @@ public abstract class AbstractKeyValueStorageTest {
storage.put(key1, val1);
// Rewrite.
- storage.put(key2, val2_1);
- storage.put(key2, val2_2);
+ storage.put(key2, val21);
+ storage.put(key2, val22);
// Remove. Tombstone must not be removed again.
- storage.put(key3, val3_1);
+ storage.put(key3, val31);
storage.remove(key3);
assertEquals(5, storage.revision());
@@ -930,10 +936,10 @@ public abstract class AbstractKeyValueStorageTest {
public void putGetRemoveCompact() {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 1);
- final byte[] val1_3 = keyValue(1, 3);
+ byte[] val13 = keyValue(1, 3);
- final byte[] key2 = key(2);
- final byte[] val2_2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val22 = keyValue(2, 2);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -958,7 +964,7 @@ public abstract class AbstractKeyValueStorageTest {
assertEquals(1, storage.updateCounter());
// Previous entry is empty.
- emptyEntry = storage.getAndPut(key2, val2_2);
+ emptyEntry = storage.getAndPut(key2, val22);
assertEquals(2, storage.revision());
assertEquals(2, storage.updateCounter());
@@ -970,14 +976,14 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e2.empty());
assertFalse(e2.tombstone());
assertArrayEquals(key2, e2.key());
- assertArrayEquals(val2_2, e2.value());
+ assertArrayEquals(val22, e2.value());
assertEquals(2, e2.revision());
assertEquals(2, e2.updateCounter());
assertEquals(2, storage.revision());
assertEquals(2, storage.updateCounter());
// Previous entry is not empty.
- e11 = storage.getAndPut(key1, val1_3);
+ e11 = storage.getAndPut(key1, val13);
assertFalse(e11.empty());
assertFalse(e11.tombstone());
@@ -994,7 +1000,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e13.empty());
assertFalse(e13.tombstone());
assertArrayEquals(key1, e13.key());
- assertArrayEquals(val1_3, e13.value());
+ assertArrayEquals(val13, e13.value());
assertEquals(3, e13.revision());
assertEquals(3, e13.updateCounter());
assertEquals(3, storage.revision());
@@ -1006,7 +1012,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e22.empty());
assertFalse(e22.tombstone());
assertArrayEquals(key2, e22.key());
- assertArrayEquals(val2_2, e22.value());
+ assertArrayEquals(val22, e22.value());
assertEquals(2, e22.revision());
assertEquals(2, e22.updateCounter());
assertEquals(4, storage.revision()); // Storage revision is changed.
@@ -1034,7 +1040,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e13.empty());
assertFalse(e13.tombstone());
assertArrayEquals(key1, e13.key());
- assertArrayEquals(val1_3, e13.value());
+ assertArrayEquals(val13, e13.value());
assertEquals(3, e13.revision());
assertEquals(3, e13.updateCounter());
assertEquals(5, storage.revision()); // Storage revision is changed.
@@ -1061,13 +1067,13 @@ public abstract class AbstractKeyValueStorageTest {
public void invokeWithRevisionCondition_successBranch() {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 11);
- final byte[] val1_2 = keyValue(1, 12);
+ byte[] val12 = keyValue(1, 12);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1079,11 +1085,8 @@ public abstract class AbstractKeyValueStorageTest {
boolean branch = storage.invoke(
new RevisionCondition(RevisionCondition.Type.EQUAL, key1, 1),
- List.of(
- new Operation(OperationType.PUT, key1, val1_2),
- new Operation(OperationType.PUT, key2, val2)
- ),
- List.of(new Operation(OperationType.PUT, key3, val3))
+ List.of(Operation.put(key1, val12), Operation.put(key2, val2)),
+ List.of(Operation.put(key3, val3))
);
// "Success" branch is applied.
@@ -1097,7 +1100,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e1.tombstone());
assertEquals(2, e1.revision());
assertEquals(2, e1.updateCounter());
- assertArrayEquals(val1_2, e1.value());
+ assertArrayEquals(val12, e1.value());
Entry e2 = storage.get(key2);
@@ -1117,13 +1120,13 @@ public abstract class AbstractKeyValueStorageTest {
public void invokeWithRevisionCondition_failureBranch() {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 11);
- final byte[] val1_2 = keyValue(1, 12);
+ byte[] val12 = keyValue(1, 12);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1135,11 +1138,8 @@ public abstract class AbstractKeyValueStorageTest {
boolean branch = storage.invoke(
new RevisionCondition(RevisionCondition.Type.EQUAL, key1, 2),
- List.of(new Operation(OperationType.PUT, key3, val3)),
- List.of(
- new Operation(OperationType.PUT, key1, val1_2),
- new Operation(OperationType.PUT, key2, val2)
- )
+ List.of(Operation.put(key3, val3)),
+ List.of(Operation.put(key1, val12), Operation.put(key2, val2))
);
// "Failure" branch is applied.
@@ -1153,7 +1153,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e1.tombstone());
assertEquals(2, e1.revision());
assertEquals(2, e1.updateCounter());
- assertArrayEquals(val1_2, e1.value());
+ assertArrayEquals(val12, e1.value());
Entry e2 = storage.get(key2);
@@ -1173,13 +1173,13 @@ public abstract class AbstractKeyValueStorageTest {
public void invokeWithExistsCondition_successBranch() {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 11);
- final byte[] val1_2 = keyValue(1, 12);
+ byte[] val12 = keyValue(1, 12);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1191,11 +1191,8 @@ public abstract class AbstractKeyValueStorageTest {
boolean branch = storage.invoke(
new ExistenceCondition(ExistenceCondition.Type.EXISTS, key1),
- List.of(
- new Operation(OperationType.PUT, key1, val1_2),
- new Operation(OperationType.PUT, key2, val2)
- ),
- List.of(new Operation(OperationType.PUT, key3, val3))
+ List.of(Operation.put(key1, val12), Operation.put(key2, val2)),
+ List.of(Operation.put(key3, val3))
);
// "Success" branch is applied.
@@ -1209,7 +1206,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e1.tombstone());
assertEquals(2, e1.revision());
assertEquals(2, e1.updateCounter());
- assertArrayEquals(val1_2, e1.value());
+ assertArrayEquals(val12, e1.value());
Entry e2 = storage.get(key2);
@@ -1229,13 +1226,13 @@ public abstract class AbstractKeyValueStorageTest {
public void invokeWithExistsCondition_failureBranch() {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 11);
- final byte[] val1_2 = keyValue(1, 12);
+ byte[] val12 = keyValue(1, 12);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1247,11 +1244,8 @@ public abstract class AbstractKeyValueStorageTest {
boolean branch = storage.invoke(
new ExistenceCondition(ExistenceCondition.Type.EXISTS, key3),
- List.of(new Operation(OperationType.PUT, key3, val3)),
- List.of(
- new Operation(OperationType.PUT, key1, val1_2),
- new Operation(OperationType.PUT, key2, val2)
- )
+ List.of(Operation.put(key3, val3)),
+ List.of(Operation.put(key1, val12), Operation.put(key2, val2))
);
// "Failure" branch is applied.
@@ -1265,7 +1259,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e1.tombstone());
assertEquals(2, e1.revision());
assertEquals(2, e1.updateCounter());
- assertArrayEquals(val1_2, e1.value());
+ assertArrayEquals(val12, e1.value());
Entry e2 = storage.get(key2);
@@ -1285,13 +1279,13 @@ public abstract class AbstractKeyValueStorageTest {
public void invokeWithNotExistsCondition_successBranch() {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 11);
- final byte[] val1_2 = keyValue(1, 12);
+ byte[] val12 = keyValue(1, 12);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1303,11 +1297,8 @@ public abstract class AbstractKeyValueStorageTest {
boolean branch = storage.invoke(
new ExistenceCondition(ExistenceCondition.Type.NOT_EXISTS, key2),
- List.of(
- new Operation(OperationType.PUT, key1, val1_2),
- new Operation(OperationType.PUT, key2, val2)
- ),
- List.of(new Operation(OperationType.PUT, key3, val3))
+ List.of(Operation.put(key1, val12), Operation.put(key2, val2)),
+ List.of(Operation.put(key3, val3))
);
// "Success" branch is applied.
@@ -1321,7 +1312,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e1.tombstone());
assertEquals(2, e1.revision());
assertEquals(2, e1.updateCounter());
- assertArrayEquals(val1_2, e1.value());
+ assertArrayEquals(val12, e1.value());
Entry e2 = storage.get(key2);
@@ -1341,13 +1332,13 @@ public abstract class AbstractKeyValueStorageTest {
public void invokeWithNotExistsCondition_failureBranch() {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 11);
- final byte[] val1_2 = keyValue(1, 12);
+ byte[] val12 = keyValue(1, 12);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1359,10 +1350,10 @@ public abstract class AbstractKeyValueStorageTest {
boolean branch = storage.invoke(
new ExistenceCondition(ExistenceCondition.Type.NOT_EXISTS, key1),
- List.of(new Operation(OperationType.PUT, key3, val3)),
+ List.of(Operation.put(key3, val3)),
List.of(
- new Operation(OperationType.PUT, key1, val1_2),
- new Operation(OperationType.PUT, key2, val2)
+ Operation.put(key1, val12),
+ Operation.put(key2, val2)
)
);
@@ -1377,7 +1368,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e1.tombstone());
assertEquals(2, e1.revision());
assertEquals(2, e1.updateCounter());
- assertArrayEquals(val1_2, e1.value());
+ assertArrayEquals(val12, e1.value());
Entry e2 = storage.get(key2);
@@ -1398,11 +1389,11 @@ public abstract class AbstractKeyValueStorageTest {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 11);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1415,8 +1406,8 @@ public abstract class AbstractKeyValueStorageTest {
boolean branch = storage.invoke(
new TombstoneCondition(key1),
- List.of(new Operation(OperationType.PUT, key2, val2)),
- List.of(new Operation(OperationType.PUT, key3, val3))
+ List.of(Operation.put(key2, val2)),
+ List.of(Operation.put(key3, val3))
);
// "Success" branch is applied.
@@ -1451,11 +1442,11 @@ public abstract class AbstractKeyValueStorageTest {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 11);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1467,8 +1458,8 @@ public abstract class AbstractKeyValueStorageTest {
boolean branch = storage.invoke(
new TombstoneCondition(key1),
- List.of(new Operation(OperationType.PUT, key2, val2)),
- List.of(new Operation(OperationType.PUT, key3, val3))
+ List.of(Operation.put(key2, val2)),
+ List.of(Operation.put(key3, val3))
);
// "Failure" branch is applied.
@@ -1502,13 +1493,13 @@ public abstract class AbstractKeyValueStorageTest {
public void invokeWithValueCondition_successBranch() {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 11);
- final byte[] val1_2 = keyValue(1, 12);
+ byte[] val12 = keyValue(1, 12);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1521,10 +1512,10 @@ public abstract class AbstractKeyValueStorageTest {
boolean branch = storage.invoke(
new ValueCondition(ValueCondition.Type.EQUAL, key1, val11),
List.of(
- new Operation(OperationType.PUT, key1, val1_2),
- new Operation(OperationType.PUT, key2, val2)
+ Operation.put(key1, val12),
+ Operation.put(key2, val2)
),
- List.of(new Operation(OperationType.PUT, key3, val3))
+ List.of(Operation.put(key3, val3))
);
// "Success" branch is applied.
@@ -1538,7 +1529,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e1.tombstone());
assertEquals(2, e1.revision());
assertEquals(2, e1.updateCounter());
- assertArrayEquals(val1_2, e1.value());
+ assertArrayEquals(val12, e1.value());
Entry e2 = storage.get(key2);
@@ -1558,13 +1549,13 @@ public abstract class AbstractKeyValueStorageTest {
public void invokeWithValueCondition_failureBranch() {
byte[] key1 = key(1);
byte[] val11 = keyValue(1, 11);
- final byte[] val1_2 = keyValue(1, 12);
+ byte[] val12 = keyValue(1, 12);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1575,11 +1566,11 @@ public abstract class AbstractKeyValueStorageTest {
assertEquals(1, storage.updateCounter());
boolean branch = storage.invoke(
- new ValueCondition(ValueCondition.Type.EQUAL, key1, val1_2),
- List.of(new Operation(OperationType.PUT, key3, val3)),
+ new ValueCondition(ValueCondition.Type.EQUAL, key1, val12),
+ List.of(Operation.put(key3, val3)),
List.of(
- new Operation(OperationType.PUT, key1, val1_2),
- new Operation(OperationType.PUT, key2, val2)
+ Operation.put(key1, val12),
+ Operation.put(key2, val2)
)
);
@@ -1594,7 +1585,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(e1.tombstone());
assertEquals(2, e1.revision());
assertEquals(2, e1.updateCounter());
- assertArrayEquals(val1_2, e1.value());
+ assertArrayEquals(val12, e1.value());
Entry e2 = storage.get(key2);
@@ -1615,11 +1606,11 @@ public abstract class AbstractKeyValueStorageTest {
byte[] key1 = key(1);
byte[] val1 = keyValue(1, 1);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1646,8 +1637,8 @@ public abstract class AbstractKeyValueStorageTest {
branch = storage.invoke(
new ValueCondition(ValueCondition.Type.EQUAL, key1, val1),
List.of(
- new Operation(OperationType.PUT, key2, val2),
- new Operation(OperationType.PUT, key3, val3)
+ Operation.put(key2, val2),
+ Operation.put(key3, val3)
),
List.of(new Operation(OperationType.NO_OP, null, null))
);
@@ -1730,12 +1721,12 @@ public abstract class AbstractKeyValueStorageTest {
byte[] val1 = keyValue(1, 1);
byte[] rval1 = keyValue(1, 4);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
- final byte[] rval3 = keyValue(2, 6);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
+ byte[] rval3 = keyValue(2, 6);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1761,13 +1752,13 @@ public abstract class AbstractKeyValueStorageTest {
new If(
new RevisionCondition(RevisionCondition.Type.EQUAL, key3, 3),
new Statement(
- new Update(List.of(new Operation(OperationType.PUT, key1, rval1)), new StatementResult(1))),
+ new Update(List.of(Operation.put(key1, rval1)), new StatementResult(1))),
new Statement(
new Update(
- List.of(new Operation(OperationType.PUT, key1, rval1),
+ List.of(Operation.put(key1, rval1),
new Operation(OperationType.REMOVE, key2, null)),
new StatementResult(2))))),
- new Statement(new Update(List.of(new Operation(OperationType.PUT, key3, rval3)), new StatementResult(3)))
+ new Statement(new Update(List.of(Operation.put(key3, rval3)), new StatementResult(3)))
);
StatementResult branch = storage.invoke(iif);
@@ -1810,12 +1801,12 @@ public abstract class AbstractKeyValueStorageTest {
byte[] val1 = keyValue(1, 1);
byte[] rval1 = keyValue(1, 4);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
- final byte[] rval3 = keyValue(2, 6);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
+ byte[] rval3 = keyValue(2, 6);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1843,11 +1834,11 @@ public abstract class AbstractKeyValueStorageTest {
If iif = new If(
new OrCondition(new ValueCondition(Type.EQUAL, key1, val1), new ExistenceCondition(ExistenceCondition.Type.EXISTS, key2)),
new Statement(new If(new RevisionCondition(RevisionCondition.Type.EQUAL, key3, 3),
- new Statement(new Update(List.of(new Operation(OperationType.PUT, key1, rval1)), new StatementResult(1))),
+ new Statement(new Update(List.of(Operation.put(key1, rval1)), new StatementResult(1))),
new Statement(new Update(
- List.of(new Operation(OperationType.PUT, key1, rval1), new Operation(OperationType.REMOVE, key2, null)),
+ List.of(Operation.put(key1, rval1), new Operation(OperationType.REMOVE, key2, null)),
new StatementResult(2))))),
- new Statement(new Update(List.of(new Operation(OperationType.PUT, key3, rval3)), new StatementResult(3))));
+ new Statement(new Update(List.of(Operation.put(key3, rval3)), new StatementResult(3))));
StatementResult branch = storage.invoke(iif);
@@ -1890,12 +1881,12 @@ public abstract class AbstractKeyValueStorageTest {
byte[] val1 = keyValue(1, 1);
byte[] rval1 = keyValue(1, 4);
- final byte[] key2 = key(2);
- final byte[] val2 = keyValue(2, 2);
+ byte[] key2 = key(2);
+ byte[] val2 = keyValue(2, 2);
- final byte[] key3 = key(3);
- final byte[] val3 = keyValue(3, 3);
- final byte[] rval3 = keyValue(2, 6);
+ byte[] key3 = key(3);
+ byte[] val3 = keyValue(3, 3);
+ byte[] rval3 = keyValue(2, 6);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1913,11 +1904,11 @@ public abstract class AbstractKeyValueStorageTest {
If iif = new If(
new OrCondition(new ValueCondition(Type.EQUAL, key1, val1), new ExistenceCondition(ExistenceCondition.Type.EXISTS, key2)),
new Statement(new If(new RevisionCondition(RevisionCondition.Type.EQUAL, key3, 3),
- new Statement(new Update(List.of(new Operation(OperationType.PUT, key1, rval1)), new StatementResult(1))),
+ new Statement(new Update(List.of(Operation.put(key1, rval1)), new StatementResult(1))),
new Statement(new Update(
- List.of(new Operation(OperationType.PUT, key1, rval1), new Operation(OperationType.REMOVE, key2, null)),
+ List.of(Operation.put(key1, rval1), new Operation(OperationType.REMOVE, key2, null)),
new StatementResult(2))))),
- new Statement(new Update(List.of(new Operation(OperationType.PUT, key3, rval3)), new StatementResult(3))));
+ new Statement(new Update(List.of(Operation.put(key3, rval3)), new StatementResult(3))));
StatementResult branch = storage.invoke(iif);
@@ -2111,7 +2102,7 @@ public abstract class AbstractKeyValueStorageTest {
for (WatchEvent event : cur) {
assertTrue(event.single());
- Entry entry = event.entryEvent().entry();
+ Entry entry = event.entryEvent().newEntry();
byte[] entryKey = entry.key();
@@ -2129,14 +2120,14 @@ public abstract class AbstractKeyValueStorageTest {
@Test
public void watchCursorForRange() throws Exception {
byte[] key1 = key(1);
- final byte[] val1_1 = keyValue(1, 11);
+ byte[] val11 = keyValue(1, 11);
- final byte[] key2 = key(2);
- final byte[] val2_1 = keyValue(2, 21);
- final byte[] val2_2 = keyValue(2, 22);
+ byte[] key2 = key(2);
+ byte[] val21 = keyValue(2, 21);
+ byte[] val22 = keyValue(2, 22);
- final byte[] key3 = key(3);
- final byte[] val3_1 = keyValue(3, 31);
+ byte[] key3 = key(3);
+ byte[] val31 = keyValue(3, 31);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -2149,7 +2140,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(it.hasNext());
assertThrows(NoSuchElementException.class, it::next);
- storage.putAll(List.of(key1, key2), List.of(val1_1, val2_1));
+ storage.putAll(List.of(key1, key2), List.of(val11, val21));
assertEquals(1, storage.revision());
assertEquals(2, storage.updateCounter());
@@ -2158,7 +2149,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(it.hasNext());
assertThrows(NoSuchElementException.class, it::next);
- storage.putAll(List.of(key2, key3), List.of(val2_2, val3_1));
+ storage.putAll(List.of(key2, key3), List.of(val22, val31));
assertEquals(2, storage.revision());
assertEquals(4, storage.updateCounter());
@@ -2171,7 +2162,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(watchEvent.single());
Map<ByteArray, EntryEvent> map = watchEvent.entryEvents().stream()
- .collect(Collectors.toMap(evt -> new ByteArray(evt.entry().key()), identity()));
+ .collect(Collectors.toMap(evt -> new ByteArray(evt.newEntry().key()), identity()));
assertEquals(2, map.size());
@@ -2187,16 +2178,16 @@ public abstract class AbstractKeyValueStorageTest {
assertEquals(1, oldEntry2.revision());
assertEquals(2, oldEntry2.updateCounter());
assertArrayEquals(key2, oldEntry2.key());
- assertArrayEquals(val2_1, oldEntry2.value());
+ assertArrayEquals(val21, oldEntry2.value());
- Entry newEntry2 = e2.entry();
+ Entry newEntry2 = e2.newEntry();
assertFalse(newEntry2.empty());
assertFalse(newEntry2.tombstone());
assertEquals(2, newEntry2.revision());
assertEquals(3, newEntry2.updateCounter());
assertArrayEquals(key2, newEntry2.key());
- assertArrayEquals(val2_2, newEntry2.value());
+ assertArrayEquals(val22, newEntry2.value());
// Second update under revision.
EntryEvent e3 = map.get(new ByteArray(key3));
@@ -2209,14 +2200,14 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(oldEntry3.tombstone());
assertArrayEquals(key3, oldEntry3.key());
- Entry newEntry3 = e3.entry();
+ Entry newEntry3 = e3.newEntry();
assertFalse(newEntry3.empty());
assertFalse(newEntry3.tombstone());
assertEquals(2, newEntry3.revision());
assertEquals(4, newEntry3.updateCounter());
assertArrayEquals(key3, newEntry3.key());
- assertArrayEquals(val3_1, newEntry3.value());
+ assertArrayEquals(val31, newEntry3.value());
assertFalse(it.hasNext());
@@ -2237,9 +2228,9 @@ public abstract class AbstractKeyValueStorageTest {
assertEquals(1, oldEntry1.revision());
assertEquals(1, oldEntry1.updateCounter());
assertArrayEquals(key1, oldEntry1.key());
- assertArrayEquals(val1_1, oldEntry1.value());
+ assertArrayEquals(val11, oldEntry1.value());
- Entry newEntry1 = e1.entry();
+ Entry newEntry1 = e1.newEntry();
assertFalse(newEntry1.empty());
assertTrue(newEntry1.tombstone());
@@ -2256,12 +2247,12 @@ public abstract class AbstractKeyValueStorageTest {
@Test
public void watchCursorForKey() {
byte[] key1 = key(1);
- final byte[] val1_1 = keyValue(1, 11);
- final byte[] val1_2 = keyValue(1, 12);
+ byte[] val11 = keyValue(1, 11);
+ byte[] val12 = keyValue(1, 12);
- final byte[] key2 = key(2);
- final byte[] val2_1 = keyValue(2, 21);
- final byte[] val2_2 = keyValue(2, 22);
+ byte[] key2 = key(2);
+ byte[] val21 = keyValue(2, 21);
+ byte[] val22 = keyValue(2, 22);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -2273,7 +2264,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(it.hasNext());
assertThrows(NoSuchElementException.class, it::next);
- storage.putAll(List.of(key1, key2), List.of(val1_1, val2_1));
+ storage.putAll(List.of(key1, key2), List.of(val11, val21));
assertEquals(1, storage.revision());
assertEquals(2, storage.updateCounter());
@@ -2291,22 +2282,22 @@ public abstract class AbstractKeyValueStorageTest {
assertTrue(oldEntry1.empty());
assertFalse(oldEntry1.tombstone());
- Entry newEntry1 = e1.entry();
+ Entry newEntry1 = e1.newEntry();
assertFalse(newEntry1.empty());
assertFalse(newEntry1.tombstone());
assertEquals(1, newEntry1.revision());
assertEquals(1, newEntry1.updateCounter());
assertArrayEquals(key1, newEntry1.key());
- assertArrayEquals(val1_1, newEntry1.value());
+ assertArrayEquals(val11, newEntry1.value());
assertFalse(it.hasNext());
- storage.put(key2, val2_2);
+ storage.put(key2, val22);
assertFalse(it.hasNext());
- storage.put(key1, val1_2);
+ storage.put(key1, val12);
assertTrue(it.hasNext());
@@ -2323,16 +2314,16 @@ public abstract class AbstractKeyValueStorageTest {
assertEquals(1, oldEntry1.revision());
assertEquals(1, oldEntry1.updateCounter());
assertArrayEquals(key1, newEntry1.key());
- assertArrayEquals(val1_1, newEntry1.value());
+ assertArrayEquals(val11, newEntry1.value());
- newEntry1 = e1.entry();
+ newEntry1 = e1.newEntry();
assertFalse(newEntry1.empty());
assertFalse(newEntry1.tombstone());
assertEquals(3, newEntry1.revision());
assertEquals(4, newEntry1.updateCounter());
assertArrayEquals(key1, newEntry1.key());
- assertArrayEquals(val1_2, newEntry1.value());
+ assertArrayEquals(val12, newEntry1.value());
assertFalse(it.hasNext());
}
@@ -2381,7 +2372,7 @@ public abstract class AbstractKeyValueStorageTest {
assertTrue(oldEntry1.empty());
assertFalse(oldEntry1.tombstone());
- Entry newEntry1 = e1.entry();
+ Entry newEntry1 = e1.newEntry();
assertFalse(newEntry1.empty());
assertFalse(newEntry1.tombstone());
@@ -2395,15 +2386,15 @@ public abstract class AbstractKeyValueStorageTest {
@Test
public void watchCursorForKeys() {
byte[] key1 = key(1);
- final byte[] val1_1 = keyValue(1, 11);
+ byte[] val11 = keyValue(1, 11);
byte[] key2 = key(2);
- final byte[] val2_1 = keyValue(2, 21);
- final byte[] val2_2 = keyValue(2, 22);
+ byte[] val21 = keyValue(2, 21);
+ byte[] val22 = keyValue(2, 22);
- final byte[] key3 = key(3);
- final byte[] val3_1 = keyValue(3, 31);
- final byte[] val3_2 = keyValue(3, 32);
+ byte[] key3 = key(3);
+ byte[] val31 = keyValue(3, 31);
+ byte[] val32 = keyValue(3, 32);
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -2415,7 +2406,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(it.hasNext());
assertThrows(NoSuchElementException.class, it::next);
- storage.putAll(List.of(key1, key2, key3), List.of(val1_1, val2_1, val3_1));
+ storage.putAll(List.of(key1, key2, key3), List.of(val11, val21, val31));
assertEquals(1, storage.revision());
assertEquals(3, storage.updateCounter());
@@ -2428,7 +2419,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(it.hasNext());
- storage.put(key2, val2_2);
+ storage.put(key2, val22);
assertTrue(it.hasNext());
@@ -2438,7 +2429,7 @@ public abstract class AbstractKeyValueStorageTest {
assertFalse(it.hasNext());
- storage.put(key3, val3_2);
+ storage.put(key3, val32);
assertFalse(it.hasNext());
}
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AndConditionTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AndConditionTest.java
index 30fd1fec2e..8fdbe9cfac 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AndConditionTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AndConditionTest.java
@@ -27,6 +27,8 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Arrays;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.impl.EntryImpl;
import org.apache.ignite.internal.util.ArrayUtils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -42,11 +44,10 @@ public class AndConditionTest {
private Condition cond3;
private Condition cond4;
- private final Entry[] entries = new Entry[] {
- new Entry(new byte[]{1}, new byte[]{10}, 1, 1),
- new Entry(new byte[]{2}, new byte[]{20}, 2, 3),
- new Entry(new byte[]{3}, new byte[]{30}, 3, 4),
-
+ private final Entry[] entries = {
+ new EntryImpl(new byte[]{1}, new byte[]{10}, 1, 1),
+ new EntryImpl(new byte[]{2}, new byte[]{20}, 2, 3),
+ new EntryImpl(new byte[]{3}, new byte[]{30}, 3, 4),
};
@BeforeEach
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/ExistenceConditionTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/ExistenceConditionTest.java
index 03ee7c4b56..6986f8e290 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/ExistenceConditionTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/ExistenceConditionTest.java
@@ -22,6 +22,8 @@ import static org.apache.ignite.internal.metastorage.server.ExistenceCondition.T
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.impl.EntryImpl;
import org.junit.jupiter.api.Test;
/**
@@ -37,13 +39,13 @@ public class ExistenceConditionTest {
private static final byte[] VAL = new byte[]{1};
/** Regular entry. */
- private static final Entry ENTRY = new Entry(KEY, VAL, 1, 1);
+ private static final Entry ENTRY = new EntryImpl(KEY, VAL, 1, 1);
/** Empty entry. */
- private static final Entry EMPTY = Entry.empty(KEY);
+ private static final Entry EMPTY = EntryImpl.empty(KEY);
/** Tombstone entry. */
- private static final Entry TOMBSTONE = Entry.tombstone(KEY, 1, 1);
+ private static final Entry TOMBSTONE = EntryImpl.tombstone(KEY, 1, 1);
/**
* Tests {@link ExistenceCondition.Type#EXISTS} condition for regular, empty and tombstone entries.
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/OrConditionTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/OrConditionTest.java
index fd403101c9..77503ed68e 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/OrConditionTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/OrConditionTest.java
@@ -27,6 +27,8 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Arrays;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.impl.EntryImpl;
import org.apache.ignite.internal.util.ArrayUtils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -42,11 +44,10 @@ public class OrConditionTest {
private Condition cond3;
private Condition cond4;
- private final Entry[] entries = new Entry[] {
- new Entry(new byte[]{1}, new byte[]{10}, 1, 1),
- new Entry(new byte[]{2}, new byte[]{20}, 2, 3),
- new Entry(new byte[]{3}, new byte[]{30}, 3, 4),
-
+ private final Entry[] entries = {
+ new EntryImpl(new byte[]{1}, new byte[]{10}, 1, 1),
+ new EntryImpl(new byte[]{2}, new byte[]{20}, 2, 3),
+ new EntryImpl(new byte[]{3}, new byte[]{30}, 3, 4),
};
@BeforeEach
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/RevisionConditionTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/RevisionConditionTest.java
index d7e906067a..1e5d9866fb 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/RevisionConditionTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/RevisionConditionTest.java
@@ -25,6 +25,7 @@ import static org.apache.ignite.internal.metastorage.server.RevisionCondition.Ty
import static org.apache.ignite.internal.metastorage.server.RevisionCondition.Type.NOT_EQUAL;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import org.apache.ignite.internal.metastorage.impl.EntryImpl;
import org.junit.jupiter.api.Test;
/**
@@ -47,7 +48,7 @@ public class RevisionConditionTest {
Condition cond = new RevisionCondition(EQUAL, KEY, 1);
// 1 == 1.
- assertTrue(cond.test(new Entry(KEY, VAL, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL, 1, 1)));
}
/**
@@ -58,7 +59,7 @@ public class RevisionConditionTest {
Condition cond = new RevisionCondition(NOT_EQUAL, KEY, 1);
// 2 != 1.
- assertTrue(cond.test(new Entry(KEY, VAL, 2, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL, 2, 1)));
}
/**
@@ -69,7 +70,7 @@ public class RevisionConditionTest {
Condition cond = new RevisionCondition(GREATER, KEY, 1);
// 2 > 1.
- assertTrue(cond.test(new Entry(KEY, VAL, 2, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL, 2, 1)));
}
/**
@@ -80,10 +81,10 @@ public class RevisionConditionTest {
Condition cond = new RevisionCondition(GREATER_OR_EQUAL, KEY, 1);
// 2 >= 1 (2 > 1).
- assertTrue(cond.test(new Entry(KEY, VAL, 2, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL, 2, 1)));
// 1 >= 1 (1 == 1).
- assertTrue(cond.test(new Entry(KEY, VAL, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL, 1, 1)));
}
/**
@@ -94,7 +95,7 @@ public class RevisionConditionTest {
Condition cond = new RevisionCondition(LESS, KEY, 2);
// 1 < 2
- assertTrue(cond.test(new Entry(KEY, VAL, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL, 1, 1)));
}
/**
@@ -105,9 +106,9 @@ public class RevisionConditionTest {
Condition cond = new RevisionCondition(LESS_OR_EQUAL, KEY, 2);
// 1 <= 2 (1 < 2)
- assertTrue(cond.test(new Entry(KEY, VAL, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL, 1, 1)));
// 1 <= 1 (1 == 1).
- assertTrue(cond.test(new Entry(KEY, VAL, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL, 1, 1)));
}
}
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/TombstoneConditionTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/TombstoneConditionTest.java
index 67e38dcb35..35f82b103c 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/TombstoneConditionTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/TombstoneConditionTest.java
@@ -20,6 +20,8 @@ package org.apache.ignite.internal.metastorage.server;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.impl.EntryImpl;
import org.junit.jupiter.api.Test;
/**
@@ -35,13 +37,13 @@ public class TombstoneConditionTest {
private static final byte[] VAL = new byte[]{1};
/** Regular entry. */
- private static final Entry ENTRY = new Entry(KEY, VAL, 1, 1);
+ private static final Entry ENTRY = new EntryImpl(KEY, VAL, 1, 1);
/** Empty entry. */
- private static final Entry EMPTY = Entry.empty(KEY);
+ private static final Entry EMPTY = EntryImpl.empty(KEY);
/** Tombstone entry. */
- private static final Entry TOMBSTONE = Entry.tombstone(KEY, 1, 1);
+ private static final Entry TOMBSTONE = EntryImpl.tombstone(KEY, 1, 1);
/**
* Tests {@link TombstoneCondition} condition for regular, empty and tombstone entries.
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/ValueConditionTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/ValueConditionTest.java
index ba5df54710..cf3dbd5004 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/ValueConditionTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/ValueConditionTest.java
@@ -26,6 +26,7 @@ import static org.apache.ignite.internal.metastorage.server.ValueCondition.Type.
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import org.apache.ignite.internal.metastorage.impl.EntryImpl;
import org.junit.jupiter.api.Test;
/**
@@ -53,9 +54,9 @@ public class ValueConditionTest {
public void eq() {
Condition cond = new ValueCondition(EQUAL, KEY, VAL_1);
- assertTrue(cond.test(new Entry(KEY, VAL_1, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL_1, 1, 1)));
- assertFalse(cond.test(new Entry(KEY, VAL_2, 1, 1)));
+ assertFalse(cond.test(new EntryImpl(KEY, VAL_2, 1, 1)));
}
/**
@@ -65,9 +66,9 @@ public class ValueConditionTest {
public void ne() {
Condition cond = new ValueCondition(NOT_EQUAL, KEY, VAL_1);
- assertTrue(cond.test(new Entry(KEY, VAL_2, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL_2, 1, 1)));
- assertFalse(cond.test(new Entry(KEY, VAL_1, 1, 1)));
+ assertFalse(cond.test(new EntryImpl(KEY, VAL_1, 1, 1)));
}
/**
@@ -78,10 +79,10 @@ public class ValueConditionTest {
Condition cond = new ValueCondition(GREATER, KEY, VAL_1);
// byte[]{22} > byte[]{11}.
- assertTrue(cond.test(new Entry(KEY, VAL_2, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL_2, 1, 1)));
// byte[]{11} > byte[]{11}.
- assertFalse(cond.test(new Entry(KEY, VAL_1, 1, 1)));
+ assertFalse(cond.test(new EntryImpl(KEY, VAL_1, 1, 1)));
}
/**
@@ -92,13 +93,13 @@ public class ValueConditionTest {
Condition cond = new ValueCondition(GREATER_OR_EQUAL, KEY, VAL_2);
// byte[]{33} >= byte[]{22}.
- assertTrue(cond.test(new Entry(KEY, VAL_3, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL_3, 1, 1)));
// byte[]{22} >= byte[]{22}.
- assertTrue(cond.test(new Entry(KEY, VAL_2, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL_2, 1, 1)));
// byte[]{11} >= byte[]{22}.
- assertFalse(cond.test(new Entry(KEY, VAL_1, 1, 1)));
+ assertFalse(cond.test(new EntryImpl(KEY, VAL_1, 1, 1)));
}
/**
@@ -109,10 +110,10 @@ public class ValueConditionTest {
Condition cond = new ValueCondition(LESS, KEY, VAL_2);
// byte[]{11} < byte[]{22}
- assertTrue(cond.test(new Entry(KEY, VAL_1, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL_1, 1, 1)));
// byte[]{22} < byte[]{22}
- assertFalse(cond.test(new Entry(KEY, VAL_2, 1, 1)));
+ assertFalse(cond.test(new EntryImpl(KEY, VAL_2, 1, 1)));
}
/**
@@ -123,12 +124,12 @@ public class ValueConditionTest {
Condition cond = new ValueCondition(LESS_OR_EQUAL, KEY, VAL_2);
// byte[]{11} <= byte[]{22}
- assertTrue(cond.test(new Entry(KEY, VAL_1, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL_1, 1, 1)));
// byte[]{22} <= byte[]{22}
- assertTrue(cond.test(new Entry(KEY, VAL_2, 1, 1)));
+ assertTrue(cond.test(new EntryImpl(KEY, VAL_2, 1, 1)));
// byte[]{33} <= byte[]{22}
- assertFalse(cond.test(new Entry(KEY, VAL_3, 1, 1)));
+ assertFalse(cond.test(new EntryImpl(KEY, VAL_3, 1, 1)));
}
}
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/watch/WatchAggregatorTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/watch/WatchAggregatorTest.java
index 1517c385ac..4d23ca00d2 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/watch/WatchAggregatorTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/watch/WatchAggregatorTest.java
@@ -274,8 +274,8 @@ public class WatchAggregatorTest {
return new Entry() {
/** {@inheritDoc} */
@Override
- public @NotNull ByteArray key() {
- return new ByteArray(key);
+ public @NotNull byte[] key() {
+ return key.getBytes(StandardCharsets.UTF_8);
}
/** {@inheritDoc} */
diff --git a/modules/metastorage/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java b/modules/metastorage/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java
index 4fded4c886..cb94d03faf 100644
--- a/modules/metastorage/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java
+++ b/modules/metastorage/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java
@@ -34,7 +34,13 @@ import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
import java.util.function.Predicate;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.EntryEvent;
+import org.apache.ignite.internal.metastorage.WatchEvent;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.metastorage.dsl.StatementResult;
import org.apache.ignite.internal.metastorage.exceptions.MetaStorageException;
+import org.apache.ignite.internal.metastorage.impl.EntryImpl;
import org.apache.ignite.internal.util.Cursor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -495,7 +501,7 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
List<Long> revs = keysIdx.get(key);
if (revs == null || revs.isEmpty()) {
- return Entry.empty(key);
+ return EntryImpl.empty(key);
}
long lastRev;
@@ -508,7 +514,7 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
// lastRev can be -1 if maxRevision return -1.
if (lastRev == -1) {
- return Entry.empty(key);
+ return EntryImpl.empty(key);
}
return doGetValue(key, lastRev);
@@ -539,22 +545,22 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
@NotNull
private Entry doGetValue(byte[] key, long lastRev) {
if (lastRev == 0) {
- return Entry.empty(key);
+ return EntryImpl.empty(key);
}
NavigableMap<byte[], Value> lastRevVals = revsIdx.get(lastRev);
if (lastRevVals == null || lastRevVals.isEmpty()) {
- return Entry.empty(key);
+ return EntryImpl.empty(key);
}
Value lastVal = lastRevVals.get(key);
if (lastVal.tombstone()) {
- return Entry.tombstone(key, lastRev, lastVal.updateCounter());
+ return EntryImpl.tombstone(key, lastRev, lastVal.updateCounter());
}
- return new Entry(key, lastVal.bytes(), lastRev, lastVal.updateCounter());
+ return new EntryImpl(key, lastVal.bytes(), lastRev, lastVal.updateCounter());
}
private long doPut(byte[] key, byte[] bytes, long curRev) {
@@ -854,9 +860,9 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
Entry newEntry;
if (val.tombstone()) {
- newEntry = Entry.tombstone(key, nextRetRev, val.updateCounter());
+ newEntry = EntryImpl.tombstone(key, nextRetRev, val.updateCounter());
} else {
- newEntry = new Entry(key, val.bytes(), nextRetRev, val.updateCounter());
+ newEntry = new EntryImpl(key, val.bytes(), nextRetRev, val.updateCounter());
}
Entry oldEntry = doGet(key, nextRetRev - 1);
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorage.java b/modules/runner/src/main/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorage.java
index 20af2e2d19..5c3b27e9d9 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorage.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorage.java
@@ -17,9 +17,11 @@
package org.apache.ignite.internal.configuration.storage;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.concurrent.CompletableFuture.supplyAsync;
import java.io.Serializable;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -39,6 +41,7 @@ import org.apache.ignite.internal.metastorage.EntryEvent;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.WatchEvent;
import org.apache.ignite.internal.metastorage.WatchListener;
+import org.apache.ignite.internal.metastorage.dsl.ConditionType;
import org.apache.ignite.internal.metastorage.dsl.Conditions;
import org.apache.ignite.internal.metastorage.dsl.Operation;
import org.apache.ignite.internal.metastorage.dsl.Operations;
@@ -106,8 +109,8 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
*
* <p>This is true for all cases except for node restart. Key-specific revision values are lost on local vault copy after restart, so
* stored {@link MetaStorageManagerImpl#APPLIED_REV} value is used instead. This fact has very important side effect: it's no longer
- * possible to use {@link SimpleCondition.RevisionCondition#eq} on {@link #MASTER_KEY}
- * in {@link DistributedConfigurationStorage#write(Map, long)}. {@link SimpleCondition.RevisionCondition#le(long)} must be used instead.
+ * possible to use {@link ConditionType#REV_EQUAL} on {@link #MASTER_KEY}
+ * in {@link DistributedConfigurationStorage#write(Map, long)}. {@link ConditionType#REV_LESS_OR_EQUAL} must be used instead.
*
* @see #MASTER_KEY
* @see MetaStorageManagerImpl#APPLIED_REV
@@ -150,7 +153,7 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
try (Cursor<Entry> entries = metaStorageMgr.range(rangeStart, rangeEnd)) {
for (Entry entry : entries) {
- ByteArray key = entry.key();
+ byte[] key = entry.key();
byte[] value = entry.value();
if (entry.tombstone()) {
@@ -160,11 +163,11 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
// Meta Storage should not return nulls as values
assert value != null;
- if (key.equals(MASTER_KEY)) {
+ if (Arrays.equals(key, MASTER_KEY.bytes())) {
continue;
}
- String dataKey = key.toString().substring(DISTRIBUTED_PREFIX.length());
+ String dataKey = new String(key, UTF_8).substring(DISTRIBUTED_PREFIX.length());
data.put(dataKey, ConfigurationSerializationUtil.fromBytes(value));
}
@@ -291,7 +294,7 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
/** {@inheritDoc} */
@Override
- public synchronized void registerConfigurationListener(@NotNull ConfigurationStorageListener lsnr) {
+ public synchronized void registerConfigurationListener(ConfigurationStorageListener lsnr) {
if (this.lsnr == null) {
this.lsnr = lsnr;
@@ -307,10 +310,10 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
for (EntryEvent event : events.entryEvents()) {
Entry e = event.newEntry();
- if (e.key().equals(MASTER_KEY)) {
+ if (Arrays.equals(e.key(), MASTER_KEY.bytes())) {
masterKeyEntry = e;
} else {
- String key = e.key().toString().substring(DISTRIBUTED_PREFIX.length());
+ String key = new String(e.key(), UTF_8).substring(DISTRIBUTED_PREFIX.length());
Serializable value = e.value() == null ? null : ConfigurationSerializationUtil.fromBytes(e.value());
diff --git a/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationCatchUpTest.java b/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationCatchUpTest.java
index 08c7d497bb..8ba3e7541e 100644
--- a/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationCatchUpTest.java
+++ b/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationCatchUpTest.java
@@ -184,7 +184,7 @@ public class DistributedConfigurationCatchUpTest {
private void setup() {
// Returns current master key revision.
when(mock.get(MASTER_KEY)).then(invocation -> {
- return CompletableFuture.completedFuture(new EntryImpl(MASTER_KEY, null, masterKeyRevision, -1));
+ return CompletableFuture.completedFuture(new EntryImpl(MASTER_KEY.bytes(), null, masterKeyRevision, -1));
});
// On any invocation - trigger storage listener.
@@ -214,7 +214,7 @@ public class DistributedConfigurationCatchUpTest {
* Triggers MetaStorage listener incrementing master key revision.
*/
private void triggerStorageListener() {
- EntryEvent entryEvent = new EntryEvent(null, new EntryImpl(MASTER_KEY, null, ++masterKeyRevision, -1));
+ EntryEvent entryEvent = new EntryEvent(null, new EntryImpl(MASTER_KEY.bytes(), null, ++masterKeyRevision, -1));
lsnr.onUpdate(new WatchEvent(entryEvent));
}
diff --git a/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorageTest.java b/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorageTest.java
index 9112c20b71..1b76c29845 100644
--- a/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorageTest.java
+++ b/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorageTest.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.configuration.storage;
-import static java.util.stream.Collectors.toList;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyCollection;
import static org.mockito.Mockito.mock;
@@ -25,23 +24,18 @@ import static org.mockito.Mockito.when;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.dsl.Operation;
-import org.apache.ignite.internal.metastorage.dsl.OperationType;
import org.apache.ignite.internal.metastorage.dsl.SimpleCondition;
import org.apache.ignite.internal.metastorage.server.Condition;
import org.apache.ignite.internal.metastorage.server.ExistenceCondition;
import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
import org.apache.ignite.internal.metastorage.server.RevisionCondition;
import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
-import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.vault.inmemory.InMemoryVaultService;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.NodeStoppingException;
-import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -92,11 +86,7 @@ public class DistributedConfigurationStorageTest extends ConfigurationStorageTes
Collection<Operation> success = invocation.getArgument(1);
Collection<Operation> failure = invocation.getArgument(2);
- boolean invokeResult = metaStorage.invoke(
- toServerCondition(condition),
- success.stream().map(DistributedConfigurationStorageTest::toServerOperation).collect(toList()),
- failure.stream().map(DistributedConfigurationStorageTest::toServerOperation).collect(toList())
- );
+ boolean invokeResult = metaStorage.invoke(toServerCondition(condition), success, failure);
return CompletableFuture.completedFuture(invokeResult);
});
@@ -106,7 +96,7 @@ public class DistributedConfigurationStorageTest extends ConfigurationStorageTes
ByteArray keyFrom = invocation.getArgument(0);
ByteArray keyTo = invocation.getArgument(1);
- return new CursorAdapter(metaStorage.range(keyFrom.bytes(), keyTo == null ? null : keyTo.bytes(), false));
+ return metaStorage.range(keyFrom.bytes(), keyTo == null ? null : keyTo.bytes(), false);
});
} catch (NodeStoppingException e) {
throw new RuntimeException(e);
@@ -123,111 +113,16 @@ public class DistributedConfigurationStorageTest extends ConfigurationStorageTes
case REV_LESS_OR_EQUAL:
return new RevisionCondition(
RevisionCondition.Type.LESS_OR_EQUAL,
- condition.inner().key(),
- ((SimpleCondition.RevisionCondition) condition.inner()).revision()
+ condition.key(),
+ ((SimpleCondition.RevisionCondition) condition).revision()
);
case KEY_NOT_EXISTS:
return new ExistenceCondition(
ExistenceCondition.Type.NOT_EXISTS,
- condition.inner().key()
+ condition.key()
);
default:
throw new UnsupportedOperationException("Unsupported condition type: " + condition.type());
}
}
-
- /**
- * Converts a {@link Operation} to a {@link org.apache.ignite.internal.metastorage.server.Operation}.
- */
- private static org.apache.ignite.internal.metastorage.server.Operation toServerOperation(Operation operation) {
- switch (operation.type()) {
- case PUT:
- return new org.apache.ignite.internal.metastorage.server.Operation(
- OperationType.PUT,
- operation.inner().key(),
- ((Operation.PutOp) (operation.inner())).value()
- );
- case REMOVE:
- return new org.apache.ignite.internal.metastorage.server.Operation(
- OperationType.REMOVE,
- operation.inner().key(),
- null
- );
- case NO_OP:
- return new org.apache.ignite.internal.metastorage.server.Operation(
- OperationType.NO_OP,
- null,
- null
- );
- default:
- throw new UnsupportedOperationException("Unsupported operation type: " + operation.type());
- }
- }
-
- /**
- * {@code Cursor} that converts {@link Entry} to {@link org.apache.ignite.internal.metastorage.server.Entry}.
- */
- private static class CursorAdapter implements Cursor<Entry> {
- /** Internal cursor. */
- private final Cursor<org.apache.ignite.internal.metastorage.server.Entry> internalCursor;
-
- /**
- * Constructor.
- *
- * @param internalCursor internal cursor.
- */
- CursorAdapter(Cursor<org.apache.ignite.internal.metastorage.server.Entry> internalCursor) {
- this.internalCursor = internalCursor;
- }
-
- /** {@inheritDoc} */
- @Override
- public void close() {
- internalCursor.close();
- }
-
- /** {@inheritDoc} */
- @Override
- public boolean hasNext() {
- return internalCursor.hasNext();
- }
-
- /** {@inheritDoc} */
- @Override
- public Entry next() {
- org.apache.ignite.internal.metastorage.server.Entry next = internalCursor.next();
-
- return new Entry() {
- @Override
- public @NotNull ByteArray key() {
- return new ByteArray(next.key());
- }
-
- @Override
- public byte @Nullable [] value() {
- return next.value();
- }
-
- @Override
- public long revision() {
- return next.revision();
- }
-
- @Override
- public long updateCounter() {
- return next.updateCounter();
- }
-
- @Override
- public boolean empty() {
- return next.empty();
- }
-
- @Override
- public boolean tombstone() {
- return next.tombstone();
- }
- };
- }
- }
}
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
index dfe9906023..b0861ca451 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
@@ -22,10 +22,14 @@ import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
+import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -433,7 +437,7 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
int lastVer = INITIAL_SCHEMA_VERSION;
for (Entry ent : cur) {
- String key = ent.key().toString();
+ String key = new String(ent.key(), StandardCharsets.UTF_8);
int descVer = extractVerFromSchemaKey(key);
if (descVer > lastVer) {
@@ -455,19 +459,15 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
*/
private byte[] schemaByVersion(UUID tblId, int ver) {
try {
- Cursor<Entry> cur = metastorageMgr.prefix(schemaWithVerHistKey(tblId, ver));
-
- if (cur.hasNext()) {
- Entry ent = cur.next();
-
- assert !cur.hasNext();
-
- return ent.value();
- }
-
- return null;
- } catch (NodeStoppingException e) {
- throw new IgniteException(e.traceId(), e.code(), e.getMessage(), e);
+ return metastorageMgr.get(schemaWithVerHistKey(tblId, ver))
+ .thenApply(Entry::value)
+ .get(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+
+ throw new IgniteInternalException("Interrupted when getting schema from metastorage", e);
+ } catch (TimeoutException | ExecutionException e) {
+ throw new IgniteInternalException("Exception when getting schema from metastorage", e);
}
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index d341c5397e..b6b438ef80 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -1989,7 +1989,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
metaStorageMgr.registerPrefixWatch(ByteArray.fromString(ASSIGNMENTS_SWITCH_REDUCE_PREFIX), new WatchListener() {
@Override
public boolean onUpdate(@NotNull WatchEvent evt) {
- ByteArray key = evt.entryEvent().newEntry().key();
+ byte[] key = evt.entryEvent().newEntry().key();
int partitionNumber = extractPartitionNumber(key);
UUID tblId = extractTableId(key, ASSIGNMENTS_SWITCH_REDUCE_PREFIX);
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtil.java b/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtil.java
index 51d90efc72..bbf84b28e6 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtil.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtil.java
@@ -27,6 +27,7 @@ import static org.apache.ignite.internal.metastorage.dsl.Operations.ops;
import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
import static org.apache.ignite.internal.metastorage.dsl.Operations.remove;
+import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
@@ -245,7 +246,7 @@ public class RebalanceUtil {
* @param key Key.
* @return Table id.
*/
- public static UUID extractTableId(ByteArray key) {
+ public static UUID extractTableId(byte[] key) {
return extractTableId(key, "");
}
@@ -256,8 +257,8 @@ public class RebalanceUtil {
* @param prefix Key prefix.
* @return Table id.
*/
- public static UUID extractTableId(ByteArray key, String prefix) {
- String strKey = key.toString();
+ public static UUID extractTableId(byte[] key, String prefix) {
+ String strKey = new String(key, StandardCharsets.UTF_8);
return UUID.fromString(strKey.substring(prefix.length(), strKey.indexOf("_part_")));
}
@@ -268,8 +269,8 @@ public class RebalanceUtil {
* @param key Key.
* @return Partition number.
*/
- public static int extractPartitionNumber(ByteArray key) {
- var strKey = key.toString();
+ public static int extractPartitionNumber(byte[] key) {
+ var strKey = new String(key, StandardCharsets.UTF_8);
return Integer.parseInt(strKey.substring(strKey.indexOf("_part_") + "_part_".length()));
}