Support newly introduced WorkerId instead of plain NodeId in ShardSessionRecords.

Author: armiol

datastore/src/main/java/io/spine/server/storage/datastore/DsSessionStorage.java

@@ -34,6 +34,7 @@
 import com.google.cloud.datastore.TimestampValue;
 import io.spine.server.delivery.ShardIndex;
 import io.spine.server.delivery.ShardSessionRecord;
+import io.spine.server.delivery.WorkerId;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 import java.util.Iterator;
@@ -165,9 +166,11 @@ private enum Column implements MessageColumn<ShardSessionRecord> {
                                  .getOfTotal());
         }),
 
-        node((m) -> {
-            return StringValue.of(m.getPickedBy()
-                                   .getValue());
+        worker((m) -> {
+            WorkerId worker = m.getWorker();
+            String value = worker.getNodeId().getValue() + '-' + worker.getValue();
+            return StringValue.of(value);
+
         }),
 
         when_last_picked((m) -> {

datastore/src/main/java/io/spine/server/storage/datastore/DsShardedWorkRegistry.java

@@ -33,6 +33,7 @@
 import io.spine.server.delivery.ShardIndex;
 import io.spine.server.delivery.ShardProcessingSession;
 import io.spine.server.delivery.ShardSessionRecord;
+import io.spine.server.delivery.WorkerId;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 import java.util.Iterator;
@@ -79,11 +80,26 @@ public DsShardedWorkRegistry(DatastoreStorageFactory factory) {
     public synchronized Optional<ShardProcessingSession> pickUp(ShardIndex index, NodeId nodeId) {
         checkNotNull(index);
         checkNotNull(nodeId);
+
+        WorkerId worker = currentWorkerFor(nodeId);
         Optional<ShardSessionRecord> result =
-                storage.updateTransactionally(index, new UpdateNodeIfAbsent(index, nodeId));
+                storage.updateTransactionally(index, new UpdateWorkerIfAbsent(index, worker));
         return result.map(this::asSession);
     }
 
+    /**
+     * Creates a worker ID by combining the given node ID with the ID of the current Java thread,
+     * in which the execution in performed.
+     */
+    @Override
+    protected WorkerId currentWorkerFor(NodeId id) {
+        long threadId = Thread.currentThread().getId();
+        return WorkerId.newBuilder()
+                .setNodeId(id)
+                .setValue(Long.toString(threadId))
+                .vBuild();
+    }
+
     @Override
     public synchronized Iterable<ShardIndex> releaseExpiredSessions(Duration inactivityPeriod) {
         return super.releaseExpiredSessions(inactivityPeriod);
@@ -123,24 +139,24 @@ protected DsSessionStorage storage() {
     }
 
     /**
-     * Updates the {@code nodeId} for the {@link ShardSessionRecord} with the specified
+     * Updates the {@code workerId} for the {@link ShardSessionRecord} with the specified
      * {@link ShardIndex} if the record has not been picked by anyone.
      *
      * <p>If there is no such a record, creates a new record.
      */
-    private static class UpdateNodeIfAbsent implements DsSessionStorage.RecordUpdate {
+    private static class UpdateWorkerIfAbsent implements DsSessionStorage.RecordUpdate {
 
         private final ShardIndex index;
-        private final NodeId nodeToSet;
+        private final WorkerId workerToSet;
 
-        private UpdateNodeIfAbsent(ShardIndex index, NodeId set) {
+        private UpdateWorkerIfAbsent(ShardIndex index, WorkerId worker) {
             this.index = index;
-            nodeToSet = set;
+            workerToSet = worker;
         }
 
         @Override
         public Optional<ShardSessionRecord> createOrUpdate(@Nullable ShardSessionRecord previous) {
-            if (previous != null && previous.hasPickedBy()) {
+            if (previous != null && previous.hasWorker()) {
                 return Optional.empty();
             }
             ShardSessionRecord.Builder builder =
@@ -150,7 +166,7 @@ public Optional<ShardSessionRecord> createOrUpdate(@Nullable ShardSessionRecord
                     : previous.toBuilder();
 
             ShardSessionRecord updated =
-                    builder.setPickedBy(nodeToSet)
+                    builder.setWorker(workerToSet)
                            .setWhenLastPicked(currentTime())
                            .vBuild();
             return Optional.of(updated);

datastore/src/test/java/io/spine/server/storage/datastore/DsShardedWorkRegistryTest.java

@@ -36,6 +36,7 @@
 import io.spine.server.delivery.ShardSessionRecord;
 import io.spine.server.delivery.ShardedWorkRegistry;
 import io.spine.server.delivery.ShardedWorkRegistryTest;
+import io.spine.server.delivery.WorkerId;
 import io.spine.testing.server.storage.datastore.TestDatastoreStorageFactory;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -47,8 +48,6 @@
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth8.assertThat;
 import static io.spine.server.storage.datastore.given.TestShardIndex.newIndex;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
 
 @DisplayName("`DsShardedWorkRegistry` should")
 class DsShardedWorkRegistryTest extends ShardedWorkRegistryTest {
@@ -82,60 +81,62 @@ protected ShardedWorkRegistry registry() {
     @DisplayName("pick up the shard and write a corresponding record to the storage")
     void pickUp() {
         Optional<ShardProcessingSession> session = registry.pickUp(index, nodeId);
-        assertTrue(session.isPresent());
+        WorkerId expectedWorker = registry.currentWorkerFor(nodeId);
+        assertThat(session).isPresent();
         assertThat(session.get()
                           .shardIndex()).isEqualTo(index);
 
         ShardSessionRecord record = readSingleRecord(index);
         assertThat(record.getIndex()).isEqualTo(index);
-        assertThat(record.getPickedBy()).isEqualTo(nodeId);
+        assertThat(record.getWorker()).isEqualTo(expectedWorker);
     }
 
     @Test
     @DisplayName("not be able to pick up the shard if it's already picked up")
     void cannotPickUpIfTaken() {
 
         Optional<ShardProcessingSession> session = registry.pickUp(index, nodeId);
-        assertTrue(session.isPresent());
+        assertThat(session).isPresent();
 
         Optional<ShardProcessingSession> sameIdxSameNode = registry.pickUp(index, nodeId);
-        assertFalse(sameIdxSameNode.isPresent());
+        assertThat(sameIdxSameNode).isEmpty();
 
         Optional<ShardProcessingSession> sameIdxAnotherNode = registry.pickUp(index, newNode());
-        assertFalse(sameIdxAnotherNode.isPresent());
+        assertThat(sameIdxAnotherNode).isEmpty();
 
         ShardIndex anotherIdx = newIndex(24, 100);
         Optional<ShardProcessingSession> anotherIdxSameNode = registry.pickUp(anotherIdx, nodeId);
-        assertTrue(anotherIdxSameNode.isPresent());
+        assertThat(anotherIdxSameNode).isPresent();
 
         Optional<ShardProcessingSession> anotherIdxAnotherNode =
                 registry.pickUp(anotherIdx, newNode());
-        assertFalse(anotherIdxAnotherNode.isPresent());
+        assertThat(anotherIdxAnotherNode).isEmpty();
     }
 
     @Test
     @DisplayName("complete the shard session (once picked up) and make it available for picking up")
     void completeSessionAndMakeItAvailable() {
         Optional<ShardProcessingSession> optional = registry.pickUp(index, nodeId);
-        assertTrue(optional.isPresent());
+        assertThat(optional).isPresent();
 
         Timestamp whenPickedFirst = readSingleRecord(index).getWhenLastPicked();
 
         DsShardProcessingSession session = (DsShardProcessingSession) optional.get();
         session.complete();
 
         ShardSessionRecord completedRecord = readSingleRecord(index);
-        assertFalse(completedRecord.hasPickedBy());
+        assertThat(completedRecord.hasWorker()).isFalse();
 
         NodeId anotherNode = newNode();
+        WorkerId anotherWorker = registry.currentWorkerFor(anotherNode);
         Optional<ShardProcessingSession> anotherOptional = registry.pickUp(index, anotherNode);
-        assertTrue(anotherOptional.isPresent());
+        assertThat(anotherOptional).isPresent();
 
         ShardSessionRecord secondSessionRecord = readSingleRecord(index);
-        assertThat(secondSessionRecord.getPickedBy()).isEqualTo(anotherNode);
+        assertThat(secondSessionRecord.getWorker()).isEqualTo(anotherWorker);
 
         Timestamp whenPickedSecond = secondSessionRecord.getWhenLastPicked();
-        assertTrue(Timestamps.compare(whenPickedFirst, whenPickedSecond) < 0);
+        assertThat(Timestamps.compare(whenPickedFirst, whenPickedSecond) < 0).isTrue();
     }
 
     @Test