tkalkirill commented on code in PR #3189:
URL: https://github.com/apache/ignite-3/pull/3189#discussion_r1484027506
##########
modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultService.java:
##########
@@ -37,9 +36,10 @@ public interface VaultService extends ManuallyCloseable {
* Retrieves an entry for the given key.
*
* @param key Key. Cannot be {@code null}.
- * @return Future that resolves into an entry for the given key, or {@code
null} no such mapping exists.
+ * @return Entry for the given key, or {@code null} no such mapping exists.
*/
- CompletableFuture<VaultEntry> get(ByteArray key);
+ @Nullable
+ VaultEntry get(ByteArray key);
Review Comment:
Maybe apply?
##########
modules/vault/src/testFixtures/java/org/apache/ignite/internal/vault/inmemory/InMemoryVaultService.java:
##########
@@ -37,85 +36,58 @@
*/
public class InMemoryVaultService implements VaultService {
/** Map to store values. */
- private final NavigableMap<ByteArray, byte[]> storage = new TreeMap<>();
-
- /** Mutex. */
- private final Object mux = new Object();
+ private final NavigableMap<ByteArray, byte[]> storage =
synchronizedNavigableMap(new TreeMap<>());
@Override
public void start() {
// No-op.
}
- /** {@inheritDoc} */
@Override
public void close() {
// No-op.
}
- /** {@inheritDoc} */
@Override
- public CompletableFuture<VaultEntry> get(ByteArray key) {
- return supplyAsync(() -> {
- synchronized (mux) {
- byte[] value = storage.get(key);
+ public VaultEntry get(ByteArray key) {
Review Comment:
Maybe apply?
##########
modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultManager.java:
##########
@@ -65,9 +63,10 @@ public void stop() {
* See {@link VaultService#get}.
*
* @param key Key. Cannot be {@code null}.
- * @return Future that resolves into an entry for the given key, or {@code
null} if no such mapping exists.
+ * @return Entry for the given key, or {@code null} if no such mapping
exists.
*/
- public CompletableFuture<VaultEntry> get(ByteArray key) {
+ @Nullable
+ public VaultEntry get(ByteArray key) {
Review Comment:
Maybe no? =)
##########
modules/vault/src/main/java/org/apache/ignite/internal/vault/persistence/PersistentVaultService.java:
##########
@@ -112,59 +93,44 @@ public void start() {
}
}
- /** {@inheritDoc} */
@Override
public void close() {
- IgniteUtils.shutdownAndAwaitTermination(threadPool, 10,
TimeUnit.SECONDS);
-
- futureTracker.cancelInFlightFutures();
-
- RocksUtils.closeAll(options, db);
+ RocksUtils.closeAll(db, options);
}
- /** {@inheritDoc} */
@Override
- public CompletableFuture<VaultEntry> get(ByteArray key) {
- return supplyAsync(() -> {
- try {
- byte[] value = db.get(key.bytes());
-
- return value == null ? null : new VaultEntry(key, value);
- } catch (RocksDBException e) {
- throw new IgniteInternalException("Unable to read data from
RocksDB", e);
- }
- });
+ public VaultEntry get(ByteArray key) {
Review Comment:
In my opinion, it is better to leave consistency in the signature, so that
when viewing the implementation, there is no confusion that the abstraction can
return `null`, but here at first glance never.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/LowWatermark.java:
##########
@@ -119,50 +119,46 @@ public LowWatermark(
*/
public void start() {
inBusyLock(busyLock, () -> {
- vaultManager.get(LOW_WATERMARK_VAULT_KEY)
- .thenCompose(vaultEntry -> inBusyLock(busyLock, () -> {
- if (vaultEntry == null) {
- scheduleUpdateLowWatermarkBusy();
+ HybridTimestamp lowWatermark = readLowWatermarkFromVault();
- return nullCompletedFuture();
- }
+ if (lowWatermark == null) {
+ LOG.info("Previous value of the low watermark was not found,
will schedule to update it");
+
+ scheduleUpdateLowWatermarkBusy();
- HybridTimestamp lowWatermark =
ByteUtils.fromBytes(vaultEntry.value());
+ return;
+ }
- return txManager.updateLowWatermark(lowWatermark)
- .thenApply(unused -> {
- this.lowWatermark.set(lowWatermark);
+ LOG.info(
+ "Low watermark has been successfully retrieved from the
vault and is scheduled to be updated: {}",
+ lowWatermark
+ );
-
runGcAndScheduleUpdateLowWatermarkBusy(lowWatermark);
+ txManager.updateLowWatermark(lowWatermark)
+ .thenRun(() -> inBusyLock(busyLock, () -> {
+ this.lowWatermark = lowWatermark;
- return lowWatermark;
- });
+ runGcAndScheduleUpdateLowWatermarkBusy(lowWatermark);
}))
- .whenComplete((lowWatermark, throwable) -> {
- if (throwable != null) {
- if (!(throwable instanceof NodeStoppingException))
{
- LOG.error("Error getting low watermark",
throwable);
+ .whenComplete((unused, throwable) -> {
+ if (throwable != null && !(throwable instanceof
NodeStoppingException)) {
+ LOG.error("Error during the Watermark manager
start", throwable);
- failureProcessor.process(new
FailureContext(CRITICAL_ERROR, throwable));
+ failureProcessor.process(new
FailureContext(CRITICAL_ERROR, throwable));
- inBusyLock(busyLock,
this::scheduleUpdateLowWatermarkBusy);
- }
- } else {
- if (lowWatermark == null) {
- LOG.info(
- "Previous value of the low watermark
was not found, will schedule to update it"
- );
- } else {
- LOG.info(
- "Low watermark has been successfully
got from the vault and is scheduled to be updated: {}",
- lowWatermark
- );
- }
+ inBusyLock(busyLock,
this::scheduleUpdateLowWatermarkBusy);
}
});
});
}
+ @Nullable
+ private HybridTimestamp readLowWatermarkFromVault() {
Review Comment:
```suggestion
private @Nullable HybridTimestamp readLowWatermarkFromVault() {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
