> {
-
- /**
- * Retrieves the previous element or null if either the element is
- * unlinked or the first element on the deque.
- */
- T getPrevious();
-
- /**
- * Sets the previous element or null if there is no link.
- */
- void setPrevious(T prev);
-
- /**
- * Retrieves the next element or null if either the element is
- * unlinked or the last element on the deque.
- */
- T getNext();
-
- /**
- * Sets the next element or null if there is no link.
- */
- void setNext(T next);
-}
diff --git a/server/src/main/java/org/opensearch/index/store/remote/utils/cache/LinkedDeque.java b/server/src/main/java/org/opensearch/index/store/remote/utils/cache/LinkedDeque.java
deleted file mode 100644
index 4ca42a7fcf24d..0000000000000
--- a/server/src/main/java/org/opensearch/index/store/remote/utils/cache/LinkedDeque.java
+++ /dev/null
@@ -1,432 +0,0 @@
-/*
- * SPDX-License-Identifier: Apache-2.0
- *
- * The OpenSearch Contributors require contributions made to
- * this file be licensed under the Apache-2.0 license or a
- * compatible open source license.
- */
-
-package org.opensearch.index.store.remote.utils.cache;
-
-import java.util.AbstractCollection;
-import java.util.Collection;
-import java.util.Deque;
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-
-/**
- * Linked list implementation of the {@link Deque} interface where the link
- * pointers are tightly integrated with the element. Linked deques have no
- * capacity restrictions; they grow as necessary to support usage. They are not
- * thread-safe; in the absence of external synchronization, they do not support
- * concurrent access by multiple threads. Null elements are prohibited.
- *
- * Most LinkedDeque operations run in constant time by assuming that
- * the {@link Linked} parameter is associated with the deque instance. Any usage
- * that violates this assumption will result in non-deterministic behavior.
- *
- * The iterators returned by this class are not fail-fast: If
- * the deque is modified at any time after the iterator is created, the iterator
- * will be in an unknown state. Thus, in the face of concurrent modification,
- * the iterator risks arbitrary, non-deterministic behavior at an undetermined
- * time in the future.
- *
- * @param the type of elements held in this collection
- *
- * @opensearch.internal
- */
-public final class LinkedDeque> extends AbstractCollection implements Deque {
- // This class provides a doubly-linked list that is optimized for the virtual
- // machine. The first and last elements are manipulated instead of a slightly
- // more convenient sentinel element to avoid the insertion of null checks with
- // NullPointerException throws in the byte code. The links to a removed
- // element are cleared to help a generational garbage collector if the
- // discarded elements inhabit more than one generation.
-
- /**
- * Pointer to first node.
- * Invariant: (first == null && last == null) ||
- * (first.prev == null)
- */
- E first;
-
- /**
- * Pointer to last node.
- * Invariant: (first == null && last == null) ||
- * (last.next == null)
- */
- E last;
-
- /**
- * Links the element to the front of the deque so that it becomes the first
- * element.
- *
- * @param e the unlinked element
- */
- void linkFirst(final E e) {
- final E f = first;
- first = e;
-
- if (f == null) {
- last = e;
- } else {
- f.setPrevious(e);
- e.setNext(f);
- }
- }
-
- /**
- * Links the element to the back of the deque so that it becomes the last
- * element.
- *
- * @param e the unlinked element
- */
- void linkLast(final E e) {
- final E l = last;
- last = e;
-
- if (l == null) {
- first = e;
- } else {
- l.setNext(e);
- e.setPrevious(l);
- }
- }
-
- /**
- * Unlinks the non-null first element.
- */
- E unlinkFirst() {
- final E f = first;
- final E next = f.getNext();
- f.setNext(null);
-
- first = next;
- if (next == null) {
- last = null;
- } else {
- next.setPrevious(null);
- }
- return f;
- }
-
- /**
- * Unlinks the non-null last element.
- */
- E unlinkLast() {
- final E l = last;
- final E prev = l.getPrevious();
- l.setPrevious(null);
- last = prev;
- if (prev == null) {
- first = null;
- } else {
- prev.setNext(null);
- }
- return l;
- }
-
- /**
- * Unlinks the non-null element.
- */
- void unlink(E e) {
- final E prev = e.getPrevious();
- final E next = e.getNext();
-
- if (prev == null) {
- first = next;
- } else {
- prev.setNext(next);
- e.setPrevious(null);
- }
-
- if (next == null) {
- last = prev;
- } else {
- next.setPrevious(prev);
- e.setNext(null);
- }
- }
-
- @Override
- public boolean isEmpty() {
- return (first == null);
- }
-
- void checkNotEmpty() {
- if (isEmpty()) {
- throw new NoSuchElementException();
- }
- }
-
- /**
- * {@inheritDoc}
- *
- * Beware that, unlike in most collections, this method is NOT a
- * constant-time operation.
- */
- @Override
- public int size() {
- int size = 0;
- for (E e = first; e != null; e = e.getNext()) {
- size++;
- }
- return size;
- }
-
- @Override
- public void clear() {
- for (E e = first; e != null;) {
- E next = e.getNext();
- e.setPrevious(null);
- e.setNext(null);
- e = next;
- }
- first = last = null;
- }
-
- @Override
- public boolean contains(Object o) {
- return (o instanceof Linked) && contains((Linked) o);
- }
-
- // A fast-path containment check
- boolean contains(Linked e) {
- return (e.getPrevious() != null) || (e.getNext() != null) || (e == first);
- }
-
- /**
- * Moves the element to the front of the deque so that it becomes the first
- * element.
- *
- * @param e the linked element
- */
- public void moveToFront(E e) {
- if (e != first) {
- unlink(e);
- linkFirst(e);
- }
- }
-
- /**
- * Moves the element to the back of the deque so that it becomes the last
- * element.
- *
- * @param e the linked element
- */
- public void moveToBack(E e) {
- if (e != last) {
- unlink(e);
- linkLast(e);
- }
- }
-
- @Override
- public E peek() {
- return peekFirst();
- }
-
- @Override
- public E peekFirst() {
- return first;
- }
-
- @Override
- public E peekLast() {
- return last;
- }
-
- @Override
- public E getFirst() {
- checkNotEmpty();
- return peekFirst();
- }
-
- @Override
- public E getLast() {
- checkNotEmpty();
- return peekLast();
- }
-
- @Override
- public E element() {
- return getFirst();
- }
-
- @Override
- public boolean offer(E e) {
- return offerLast(e);
- }
-
- @Override
- public boolean offerFirst(E e) {
- if (contains(e)) {
- return false;
- }
- linkFirst(e);
- return true;
- }
-
- @Override
- public boolean offerLast(E e) {
- if (contains(e)) {
- return false;
- }
- linkLast(e);
- return true;
- }
-
- @Override
- public boolean add(E e) {
- return offerLast(e);
- }
-
- @Override
- public void addFirst(E e) {
- if (!offerFirst(e)) {
- throw new IllegalArgumentException();
- }
- }
-
- @Override
- public void addLast(E e) {
- if (!offerLast(e)) {
- throw new IllegalArgumentException();
- }
- }
-
- @Override
- public E poll() {
- return pollFirst();
- }
-
- @Override
- public E pollFirst() {
- return isEmpty() ? null : unlinkFirst();
- }
-
- @Override
- public E pollLast() {
- return isEmpty() ? null : unlinkLast();
- }
-
- @Override
- public E remove() {
- return removeFirst();
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public boolean remove(Object o) {
- return (o instanceof Linked) && remove((E) o);
- }
-
- // A fast-path removal
- boolean remove(E e) {
- if (contains(e)) {
- unlink(e);
- return true;
- }
- return false;
- }
-
- @Override
- public E removeFirst() {
- checkNotEmpty();
- return pollFirst();
- }
-
- @Override
- public boolean removeFirstOccurrence(Object o) {
- return remove(o);
- }
-
- @Override
- public E removeLast() {
- checkNotEmpty();
- return pollLast();
- }
-
- @Override
- public boolean removeLastOccurrence(Object o) {
- return remove(o);
- }
-
- @Override
- public boolean removeAll(Collection c) {
- boolean modified = false;
- for (Object o : c) {
- modified |= remove(o);
- }
- return modified;
- }
-
- @Override
- public void push(E e) {
- addFirst(e);
- }
-
- @Override
- public E pop() {
- return removeFirst();
- }
-
- @Override
- public Iterator iterator() {
- return new AbstractLinkedIterator(first) {
- @Override
- E computeNext() {
- return cursor.getNext();
- }
- };
- }
-
- @Override
- public Iterator descendingIterator() {
- return new AbstractLinkedIterator(last) {
- @Override
- E computeNext() {
- return cursor.getPrevious();
- }
- };
- }
-
- abstract class AbstractLinkedIterator implements Iterator {
- E cursor;
-
- /**
- * Creates an iterator that can can traverse the deque.
- *
- * @param start the initial element to begin traversal from
- */
- AbstractLinkedIterator(E start) {
- cursor = start;
- }
-
- @Override
- public boolean hasNext() {
- return (cursor != null);
- }
-
- @Override
- public E next() {
- if (!hasNext()) {
- throw new NoSuchElementException();
- }
- E e = cursor;
- cursor = computeNext();
- return e;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
-
- /**
- * Retrieves the next element to traverse to or null if there are
- * no more elements.
- */
- abstract E computeNext();
- }
-}
diff --git a/server/src/main/java/org/opensearch/indices/IndicesQueryCache.java b/server/src/main/java/org/opensearch/indices/IndicesQueryCache.java
index 78970267f67c2..2669da3f417c3 100644
--- a/server/src/main/java/org/opensearch/indices/IndicesQueryCache.java
+++ b/server/src/main/java/org/opensearch/indices/IndicesQueryCache.java
@@ -128,13 +128,17 @@ public QueryCacheStats getStats(ShardId shard) {
// We also have some shared ram usage that we try to distribute to
// proportionally to their number of cache entries of each shard
- long totalSize = 0;
- for (QueryCacheStats s : stats.values()) {
- totalSize += s.getCacheSize();
+ if (stats.isEmpty()) {
+ shardStats.add(new QueryCacheStats(sharedRamBytesUsed, 0, 0, 0, 0));
+ } else {
+ long totalSize = 0;
+ for (QueryCacheStats s : stats.values()) {
+ totalSize += s.getCacheSize();
+ }
+ final double weight = totalSize == 0 ? 1d / stats.size() : ((double) shardStats.getCacheSize()) / totalSize;
+ final long additionalRamBytesUsed = Math.round(weight * sharedRamBytesUsed);
+ shardStats.add(new QueryCacheStats(additionalRamBytesUsed, 0, 0, 0, 0));
}
- final double weight = totalSize == 0 ? 1d / stats.size() : ((double) shardStats.getCacheSize()) / totalSize;
- final long additionalRamBytesUsed = Math.round(weight * sharedRamBytesUsed);
- shardStats.add(new QueryCacheStats(additionalRamBytesUsed, 0, 0, 0, 0));
return shardStats;
}
diff --git a/server/src/main/java/org/opensearch/indices/IndicesService.java b/server/src/main/java/org/opensearch/indices/IndicesService.java
index 9c83a33a37712..bb880f2df58da 100644
--- a/server/src/main/java/org/opensearch/indices/IndicesService.java
+++ b/server/src/main/java/org/opensearch/indices/IndicesService.java
@@ -288,7 +288,7 @@ public class IndicesService extends AbstractLifecycleComponent
private final Set danglingIndicesToWrite = Sets.newConcurrentHashSet();
private final boolean nodeWriteDanglingIndicesInfo;
private final ValuesSourceRegistry valuesSourceRegistry;
- private final IndexStorePlugin.RemoteDirectoryFactory remoteDirectoryFactory;
+ private final IndexStorePlugin.DirectoryFactory remoteDirectoryFactory;
private final BiFunction translogFactorySupplier;
@Override
@@ -318,7 +318,7 @@ public IndicesService(
Map directoryFactories,
ValuesSourceRegistry valuesSourceRegistry,
Map recoveryStateFactories,
- IndexStorePlugin.RemoteDirectoryFactory remoteDirectoryFactory,
+ IndexStorePlugin.DirectoryFactory remoteDirectoryFactory,
Supplier repositoriesServiceSupplier
) {
this.settings = settings;
@@ -433,7 +433,7 @@ public IndicesService(
Map directoryFactories,
ValuesSourceRegistry valuesSourceRegistry,
Map recoveryStateFactories,
- IndexStorePlugin.RemoteDirectoryFactory remoteDirectoryFactory,
+ IndexStorePlugin.DirectoryFactory remoteDirectoryFactory,
Supplier repositoriesServiceSupplier
) {
this.settings = settings;
diff --git a/server/src/main/java/org/opensearch/indices/replication/OngoingSegmentReplications.java b/server/src/main/java/org/opensearch/indices/replication/OngoingSegmentReplications.java
index 6f04c6cf6f665..3ab0a7539fb06 100644
--- a/server/src/main/java/org/opensearch/indices/replication/OngoingSegmentReplications.java
+++ b/server/src/main/java/org/opensearch/indices/replication/OngoingSegmentReplications.java
@@ -14,6 +14,7 @@
import org.opensearch.OpenSearchException;
import org.opensearch.action.ActionListener;
import org.opensearch.cluster.node.DiscoveryNode;
+import org.opensearch.common.util.CancellableThreads;
import org.opensearch.common.util.concurrent.ConcurrentCollections;
import org.opensearch.index.IndexService;
import org.opensearch.index.shard.IndexShard;
@@ -147,6 +148,12 @@ void startSegmentCopy(GetSegmentFilesRequest request, ActionListener {
if (segrepHandler != null) {
logger.warn("Override handler for allocation id {}", request.getTargetAllocationId());
diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java
index bf626ff93760c..1858449e13ae8 100644
--- a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java
+++ b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java
@@ -401,7 +401,7 @@ public void messageReceived(final ForceSyncRequest request, TransportChannel cha
return;
}
startReplication(
- ReplicationCheckpoint.empty(request.getShardId()),
+ ReplicationCheckpoint.empty(request.getShardId(), indexShard.getDefaultCodecName()),
indexShard,
new SegmentReplicationTargetService.SegmentReplicationListener() {
@Override
diff --git a/server/src/main/java/org/opensearch/indices/replication/checkpoint/ReplicationCheckpoint.java b/server/src/main/java/org/opensearch/indices/replication/checkpoint/ReplicationCheckpoint.java
index 57e667b06a223..32521fb0cd944 100644
--- a/server/src/main/java/org/opensearch/indices/replication/checkpoint/ReplicationCheckpoint.java
+++ b/server/src/main/java/org/opensearch/indices/replication/checkpoint/ReplicationCheckpoint.java
@@ -8,6 +8,7 @@
package org.opensearch.indices.replication.checkpoint;
+import org.opensearch.Version;
import org.opensearch.common.Nullable;
import org.opensearch.common.io.stream.StreamInput;
import org.opensearch.common.io.stream.StreamOutput;
@@ -30,29 +31,32 @@ public class ReplicationCheckpoint implements Writeable, Comparable tracker.shardAllocationId.equals(id) == false)
.collect(Collectors.toSet());
- final ReplicationCheckpoint initialCheckpoint = new ReplicationCheckpoint(tracker.shardId(), 0L, 1, 1, 1L);
- final ReplicationCheckpoint secondCheckpoint = new ReplicationCheckpoint(tracker.shardId(), 0L, 2, 2, 50L);
- final ReplicationCheckpoint thirdCheckpoint = new ReplicationCheckpoint(tracker.shardId(), 0L, 2, 3, 100L);
+ final ReplicationCheckpoint initialCheckpoint = new ReplicationCheckpoint(
+ tracker.shardId(),
+ 0L,
+ 1,
+ 1,
+ 1L,
+ Codec.getDefault().getName()
+ );
+ final ReplicationCheckpoint secondCheckpoint = new ReplicationCheckpoint(
+ tracker.shardId(),
+ 0L,
+ 2,
+ 2,
+ 50L,
+ Codec.getDefault().getName()
+ );
+ final ReplicationCheckpoint thirdCheckpoint = new ReplicationCheckpoint(
+ tracker.shardId(),
+ 0L,
+ 2,
+ 3,
+ 100L,
+ Codec.getDefault().getName()
+ );
tracker.setLatestReplicationCheckpoint(initialCheckpoint);
tracker.setLatestReplicationCheckpoint(secondCheckpoint);
diff --git a/server/src/test/java/org/opensearch/index/shard/SegmentReplicationIndexShardTests.java b/server/src/test/java/org/opensearch/index/shard/SegmentReplicationIndexShardTests.java
index 014a37249612b..c4db88782638f 100644
--- a/server/src/test/java/org/opensearch/index/shard/SegmentReplicationIndexShardTests.java
+++ b/server/src/test/java/org/opensearch/index/shard/SegmentReplicationIndexShardTests.java
@@ -8,6 +8,7 @@
package org.opensearch.index.shard;
+import org.apache.lucene.codecs.Codec;
import org.apache.lucene.index.SegmentInfos;
import org.junit.Assert;
import org.opensearch.ExceptionsHelper;
@@ -306,7 +307,7 @@ public void testRejectCheckpointOnShardRoutingPrimary() throws IOException {
assertEquals(false, primaryShard.getReplicationTracker().isPrimaryMode());
assertEquals(true, primaryShard.routingEntry().primary());
- spy.onNewCheckpoint(new ReplicationCheckpoint(primaryShard.shardId(), 0L, 0L, 0L), spyShard);
+ spy.onNewCheckpoint(new ReplicationCheckpoint(primaryShard.shardId(), 0L, 0L, 0L, Codec.getDefault().getName()), spyShard);
// Verify that checkpoint is not processed as shard routing is primary.
verify(spy, times(0)).startReplication(any(), any(), any());
@@ -1020,7 +1021,7 @@ private void assertDocCounts(IndexShard indexShard, int expectedPersistedDocCoun
private void resolveCheckpointInfoResponseListener(ActionListener listener, IndexShard primary) {
try {
- final CopyState copyState = new CopyState(ReplicationCheckpoint.empty(primary.shardId), primary);
+ final CopyState copyState = new CopyState(ReplicationCheckpoint.empty(primary.shardId, primary.getDefaultCodecName()), primary);
listener.onResponse(
new CheckpointInfoResponse(copyState.getCheckpoint(), copyState.getMetadataMap(), copyState.getInfosBytes())
);
@@ -1034,7 +1035,7 @@ private void startReplicationAndAssertCancellation(IndexShard replica, SegmentRe
throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
final SegmentReplicationTarget target = targetService.startReplication(
- ReplicationCheckpoint.empty(replica.shardId),
+ ReplicationCheckpoint.empty(replica.shardId, replica.getDefaultCodecName()),
replica,
new SegmentReplicationTargetService.SegmentReplicationListener() {
@Override
diff --git a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryFactoryTests.java b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryFactoryTests.java
index 0105d0dc309c2..7be86aa0d96a4 100644
--- a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryFactoryTests.java
+++ b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryFactoryTests.java
@@ -52,7 +52,10 @@ public void setup() {
}
public void testNewDirectory() throws IOException {
- Settings settings = Settings.builder().put(IndexMetadata.SETTING_INDEX_UUID, "uuid_1").build();
+ Settings settings = Settings.builder()
+ .put(IndexMetadata.SETTING_INDEX_UUID, "uuid_1")
+ .put(IndexMetadata.SETTING_REMOTE_STORE_REPOSITORY, "remote_store_repository")
+ .build();
IndexSettings indexSettings = IndexSettingsModule.newIndexSettings("foo", settings);
Path tempDir = createTempDir().resolve(indexSettings.getUUID()).resolve("0");
ShardPath shardPath = new ShardPath(false, tempDir, tempDir, new ShardId(indexSettings.getIndex(), 0));
@@ -66,7 +69,7 @@ public void testNewDirectory() throws IOException {
when(repositoriesService.repository("remote_store_repository")).thenReturn(repository);
- try (Directory directory = remoteSegmentStoreDirectoryFactory.newDirectory("remote_store_repository", indexSettings, shardPath)) {
+ try (Directory directory = remoteSegmentStoreDirectoryFactory.newDirectory(indexSettings, shardPath)) {
assertTrue(directory instanceof RemoteSegmentStoreDirectory);
ArgumentCaptor blobPathCaptor = ArgumentCaptor.forClass(BlobPath.class);
verify(blobStore, times(2)).blobContainer(blobPathCaptor.capture());
@@ -80,17 +83,14 @@ public void testNewDirectory() throws IOException {
}
public void testNewDirectoryRepositoryDoesNotExist() {
- Settings settings = Settings.builder().build();
+ Settings settings = Settings.builder().put(IndexMetadata.SETTING_REMOTE_STORE_REPOSITORY, "remote_store_repository").build();
IndexSettings indexSettings = IndexSettingsModule.newIndexSettings("foo", settings);
Path tempDir = createTempDir().resolve(indexSettings.getUUID()).resolve("0");
ShardPath shardPath = new ShardPath(false, tempDir, tempDir, new ShardId(indexSettings.getIndex(), 0));
when(repositoriesService.repository("remote_store_repository")).thenThrow(new RepositoryMissingException("Missing"));
- assertThrows(
- IllegalArgumentException.class,
- () -> remoteSegmentStoreDirectoryFactory.newDirectory("remote_store_repository", indexSettings, shardPath)
- );
+ assertThrows(IllegalArgumentException.class, () -> remoteSegmentStoreDirectoryFactory.newDirectory(indexSettings, shardPath));
}
}
diff --git a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java
index 956279c3ea048..49a2d50dfae06 100644
--- a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java
+++ b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java
@@ -64,16 +64,17 @@ public void testUploadedSegmentMetadataToString() {
RemoteSegmentStoreDirectory.UploadedSegmentMetadata metadata = new RemoteSegmentStoreDirectory.UploadedSegmentMetadata(
"abc",
"pqr",
- "123456"
+ "123456",
+ 1234
);
- assertEquals("abc::pqr::123456", metadata.toString());
+ assertEquals("abc::pqr::123456::1234", metadata.toString());
}
public void testUploadedSegmentMetadataFromString() {
RemoteSegmentStoreDirectory.UploadedSegmentMetadata metadata = RemoteSegmentStoreDirectory.UploadedSegmentMetadata.fromString(
- "_0.cfe::_0.cfe__uuidxyz::4567"
+ "_0.cfe::_0.cfe__uuidxyz::4567::372000"
);
- assertEquals("_0.cfe::_0.cfe__uuidxyz::4567", metadata.toString());
+ assertEquals("_0.cfe::_0.cfe__uuidxyz::4567::372000", metadata.toString());
}
public void testGetMetadataFilename() {
@@ -141,9 +142,42 @@ public void testInitNoMetadataFile() throws IOException {
private Map getDummyMetadata(String prefix, int commitGeneration) {
Map metadata = new HashMap<>();
- metadata.put(prefix + ".cfe", prefix + ".cfe::" + prefix + ".cfe__" + UUIDs.base64UUID() + "::" + randomIntBetween(1000, 5000));
- metadata.put(prefix + ".cfs", prefix + ".cfs::" + prefix + ".cfs__" + UUIDs.base64UUID() + "::" + randomIntBetween(1000, 5000));
- metadata.put(prefix + ".si", prefix + ".si::" + prefix + ".si__" + UUIDs.base64UUID() + "::" + randomIntBetween(1000, 5000));
+ metadata.put(
+ prefix + ".cfe",
+ prefix
+ + ".cfe::"
+ + prefix
+ + ".cfe__"
+ + UUIDs.base64UUID()
+ + "::"
+ + randomIntBetween(1000, 5000)
+ + "::"
+ + randomIntBetween(512000, 1024000)
+ );
+ metadata.put(
+ prefix + ".cfs",
+ prefix
+ + ".cfs::"
+ + prefix
+ + ".cfs__"
+ + UUIDs.base64UUID()
+ + "::"
+ + randomIntBetween(1000, 5000)
+ + "::"
+ + randomIntBetween(512000, 1024000)
+ );
+ metadata.put(
+ prefix + ".si",
+ prefix
+ + ".si::"
+ + prefix
+ + ".si__"
+ + UUIDs.base64UUID()
+ + "::"
+ + randomIntBetween(1000, 5000)
+ + "::"
+ + randomIntBetween(512000, 1024000)
+ );
metadata.put(
"segments_" + commitGeneration,
"segments_"
@@ -154,6 +188,8 @@ private Map getDummyMetadata(String prefix, int commitGeneration
+ UUIDs.base64UUID()
+ "::"
+ randomIntBetween(1000, 5000)
+ + "::"
+ + randomIntBetween(1024, 5120)
);
return metadata;
}
@@ -250,7 +286,7 @@ public void testDeleteFileException() throws IOException {
assertThrows(IOException.class, () -> remoteSegmentStoreDirectory.deleteFile("_0.si"));
}
- public void testFileLenght() throws IOException {
+ public void testFileLength() throws IOException {
populateMetadata();
remoteSegmentStoreDirectory.init();
@@ -259,9 +295,7 @@ public void testFileLenght() throws IOException {
assertTrue(uploadedSegments.containsKey("_0.si"));
- when(remoteDataDirectory.fileLength(startsWith("_0.si"))).thenReturn(1234L);
-
- assertEquals(1234L, remoteSegmentStoreDirectory.fileLength("_0.si"));
+ assertEquals(uploadedSegments.get("_0.si").getLength(), remoteSegmentStoreDirectory.fileLength("_0.si"));
}
public void testFileLenghtNoSuchFile() throws IOException {
@@ -376,8 +410,8 @@ public void testContainsFile() throws IOException {
);
Map metadata = new HashMap<>();
- metadata.put("_0.cfe", "_0.cfe::_0.cfe__" + UUIDs.base64UUID() + "::1234");
- metadata.put("_0.cfs", "_0.cfs::_0.cfs__" + UUIDs.base64UUID() + "::2345");
+ metadata.put("_0.cfe", "_0.cfe::_0.cfe__" + UUIDs.base64UUID() + "::1234::512");
+ metadata.put("_0.cfs", "_0.cfs::_0.cfs__" + UUIDs.base64UUID() + "::2345::1024");
when(remoteMetadataDirectory.openInput("metadata__1__5__abc", IOContext.DEFAULT)).thenReturn(createMetadataFileBytes(metadata));
@@ -390,7 +424,7 @@ public void testContainsFile() throws IOException {
UnsupportedOperationException.class,
() -> uploadedSegmentMetadataMap.put(
"_100.si",
- new RemoteSegmentStoreDirectory.UploadedSegmentMetadata("_100.si", "_100.si__uuid1", "1234")
+ new RemoteSegmentStoreDirectory.UploadedSegmentMetadata("_100.si", "_100.si__uuid1", "1234", 500)
)
);
@@ -531,8 +565,8 @@ public void testIncorrectChecksumCorruptIndexException() throws IOException {
);
Map metadata = new HashMap<>();
- metadata.put("_0.cfe", "_0.cfe::_0.cfe__" + UUIDs.base64UUID() + "::1234");
- metadata.put("_0.cfs", "_0.cfs::_0.cfs__" + UUIDs.base64UUID() + "::2345");
+ metadata.put("_0.cfe", "_0.cfe::_0.cfe__" + UUIDs.base64UUID() + "::1234::512");
+ metadata.put("_0.cfs", "_0.cfs::_0.cfs__" + UUIDs.base64UUID() + "::2345::1024");
BytesStreamOutput output = new BytesStreamOutput();
IndexOutput indexOutput = new OutputStreamIndexOutput("segment metadata", "metadata output stream", output, 4096);
diff --git a/server/src/test/java/org/opensearch/index/store/remote/metadata/RemoteSegmentMetadataHandlerTests.java b/server/src/test/java/org/opensearch/index/store/remote/metadata/RemoteSegmentMetadataHandlerTests.java
index 2a30e58b8802c..3a73015c25589 100644
--- a/server/src/test/java/org/opensearch/index/store/remote/metadata/RemoteSegmentMetadataHandlerTests.java
+++ b/server/src/test/java/org/opensearch/index/store/remote/metadata/RemoteSegmentMetadataHandlerTests.java
@@ -59,11 +59,27 @@ private Map getDummyData() {
String prefix = "_0";
expectedOutput.put(
prefix + ".cfe",
- prefix + ".cfe::" + prefix + ".cfe__" + UUIDs.base64UUID() + "::" + randomIntBetween(1000, 5000)
+ prefix
+ + ".cfe::"
+ + prefix
+ + ".cfe__"
+ + UUIDs.base64UUID()
+ + "::"
+ + randomIntBetween(1000, 5000)
+ + "::"
+ + randomIntBetween(1024, 2048)
);
expectedOutput.put(
prefix + ".cfs",
- prefix + ".cfs::" + prefix + ".cfs__" + UUIDs.base64UUID() + "::" + randomIntBetween(1000, 5000)
+ prefix
+ + ".cfs::"
+ + prefix
+ + ".cfs__"
+ + UUIDs.base64UUID()
+ + "::"
+ + randomIntBetween(1000, 5000)
+ + "::"
+ + randomIntBetween(1024, 2048)
);
return expectedOutput;
}
diff --git a/server/src/test/java/org/opensearch/indices/IndicesQueryCacheTests.java b/server/src/test/java/org/opensearch/indices/IndicesQueryCacheTests.java
index b1a44d94aa0e1..5d293e6c598f6 100644
--- a/server/src/test/java/org/opensearch/indices/IndicesQueryCacheTests.java
+++ b/server/src/test/java/org/opensearch/indices/IndicesQueryCacheTests.java
@@ -143,6 +143,7 @@ public void testBasics() throws IOException {
assertEquals(0L, stats.getCacheCount());
assertEquals(0L, stats.getHitCount());
assertEquals(0L, stats.getMissCount());
+ assertEquals(0L, stats.getMemorySizeInBytes());
assertEquals(1, s.count(new DummyQuery(0)));
@@ -151,6 +152,7 @@ public void testBasics() throws IOException {
assertEquals(1L, stats.getCacheCount());
assertEquals(0L, stats.getHitCount());
assertEquals(1L, stats.getMissCount());
+ assertTrue(stats.getMemorySizeInBytes() > 0L && stats.getMemorySizeInBytes() < Long.MAX_VALUE);
for (int i = 1; i < 20; ++i) {
assertEquals(1, s.count(new DummyQuery(i)));
@@ -161,6 +163,7 @@ public void testBasics() throws IOException {
assertEquals(20L, stats.getCacheCount());
assertEquals(0L, stats.getHitCount());
assertEquals(20L, stats.getMissCount());
+ assertTrue(stats.getMemorySizeInBytes() > 0L && stats.getMemorySizeInBytes() < Long.MAX_VALUE);
s.count(new DummyQuery(10));
@@ -169,6 +172,7 @@ public void testBasics() throws IOException {
assertEquals(20L, stats.getCacheCount());
assertEquals(1L, stats.getHitCount());
assertEquals(20L, stats.getMissCount());
+ assertTrue(stats.getMemorySizeInBytes() > 0L && stats.getMemorySizeInBytes() < Long.MAX_VALUE);
IOUtils.close(r, dir);
@@ -178,6 +182,7 @@ public void testBasics() throws IOException {
assertEquals(20L, stats.getCacheCount());
assertEquals(1L, stats.getHitCount());
assertEquals(20L, stats.getMissCount());
+ assertTrue(stats.getMemorySizeInBytes() > 0L && stats.getMemorySizeInBytes() < Long.MAX_VALUE);
cache.onClose(shard);
@@ -187,6 +192,7 @@ public void testBasics() throws IOException {
assertEquals(0L, stats.getCacheCount());
assertEquals(0L, stats.getHitCount());
assertEquals(0L, stats.getMissCount());
+ assertTrue(stats.getMemorySizeInBytes() >= 0L && stats.getMemorySizeInBytes() < Long.MAX_VALUE);
cache.close(); // this triggers some assertions
}
@@ -227,12 +233,14 @@ public void testTwoShards() throws IOException {
assertEquals(1L, stats1.getCacheCount());
assertEquals(0L, stats1.getHitCount());
assertEquals(1L, stats1.getMissCount());
+ assertTrue(stats1.getMemorySizeInBytes() >= 0L && stats1.getMemorySizeInBytes() < Long.MAX_VALUE);
QueryCacheStats stats2 = cache.getStats(shard2);
assertEquals(0L, stats2.getCacheSize());
assertEquals(0L, stats2.getCacheCount());
assertEquals(0L, stats2.getHitCount());
assertEquals(0L, stats2.getMissCount());
+ assertTrue(stats2.getMemorySizeInBytes() >= 0L && stats2.getMemorySizeInBytes() < Long.MAX_VALUE);
assertEquals(1, s2.count(new DummyQuery(0)));
@@ -241,12 +249,14 @@ public void testTwoShards() throws IOException {
assertEquals(1L, stats1.getCacheCount());
assertEquals(0L, stats1.getHitCount());
assertEquals(1L, stats1.getMissCount());
+ assertTrue(stats1.getMemorySizeInBytes() >= 0L && stats1.getMemorySizeInBytes() < Long.MAX_VALUE);
stats2 = cache.getStats(shard2);
assertEquals(1L, stats2.getCacheSize());
assertEquals(1L, stats2.getCacheCount());
assertEquals(0L, stats2.getHitCount());
assertEquals(1L, stats2.getMissCount());
+ assertTrue(stats2.getMemorySizeInBytes() >= 0L && stats2.getMemorySizeInBytes() < Long.MAX_VALUE);
for (int i = 0; i < 20; ++i) {
assertEquals(1, s2.count(new DummyQuery(i)));
@@ -257,12 +267,14 @@ public void testTwoShards() throws IOException {
assertEquals(1L, stats1.getCacheCount());
assertEquals(0L, stats1.getHitCount());
assertEquals(1L, stats1.getMissCount());
+ assertTrue(stats1.getMemorySizeInBytes() >= 0L && stats1.getMemorySizeInBytes() < Long.MAX_VALUE);
stats2 = cache.getStats(shard2);
assertEquals(10L, stats2.getCacheSize());
assertEquals(20L, stats2.getCacheCount());
assertEquals(1L, stats2.getHitCount());
assertEquals(20L, stats2.getMissCount());
+ assertTrue(stats2.getMemorySizeInBytes() >= 0L && stats2.getMemorySizeInBytes() < Long.MAX_VALUE);
IOUtils.close(r1, dir1);
@@ -272,12 +284,14 @@ public void testTwoShards() throws IOException {
assertEquals(1L, stats1.getCacheCount());
assertEquals(0L, stats1.getHitCount());
assertEquals(1L, stats1.getMissCount());
+ assertTrue(stats1.getMemorySizeInBytes() >= 0L && stats1.getMemorySizeInBytes() < Long.MAX_VALUE);
stats2 = cache.getStats(shard2);
assertEquals(10L, stats2.getCacheSize());
assertEquals(20L, stats2.getCacheCount());
assertEquals(1L, stats2.getHitCount());
assertEquals(20L, stats2.getMissCount());
+ assertTrue(stats2.getMemorySizeInBytes() >= 0L && stats2.getMemorySizeInBytes() < Long.MAX_VALUE);
cache.onClose(shard1);
@@ -287,12 +301,14 @@ public void testTwoShards() throws IOException {
assertEquals(0L, stats1.getCacheCount());
assertEquals(0L, stats1.getHitCount());
assertEquals(0L, stats1.getMissCount());
+ assertTrue(stats1.getMemorySizeInBytes() >= 0L && stats1.getMemorySizeInBytes() < Long.MAX_VALUE);
stats2 = cache.getStats(shard2);
assertEquals(10L, stats2.getCacheSize());
assertEquals(20L, stats2.getCacheCount());
assertEquals(1L, stats2.getHitCount());
assertEquals(20L, stats2.getMissCount());
+ assertTrue(stats2.getMemorySizeInBytes() >= 0L && stats2.getMemorySizeInBytes() < Long.MAX_VALUE);
IOUtils.close(r2, dir2);
cache.onClose(shard2);
@@ -303,12 +319,14 @@ public void testTwoShards() throws IOException {
assertEquals(0L, stats1.getCacheCount());
assertEquals(0L, stats1.getHitCount());
assertEquals(0L, stats1.getMissCount());
+ assertTrue(stats1.getMemorySizeInBytes() >= 0L && stats1.getMemorySizeInBytes() < Long.MAX_VALUE);
stats2 = cache.getStats(shard2);
assertEquals(0L, stats2.getCacheSize());
assertEquals(0L, stats2.getCacheCount());
assertEquals(0L, stats2.getHitCount());
assertEquals(0L, stats2.getMissCount());
+ assertTrue(stats2.getMemorySizeInBytes() >= 0L && stats2.getMemorySizeInBytes() < Long.MAX_VALUE);
cache.close(); // this triggers some assertions
}
diff --git a/server/src/test/java/org/opensearch/indices/replication/OngoingSegmentReplicationsTests.java b/server/src/test/java/org/opensearch/indices/replication/OngoingSegmentReplicationsTests.java
index 78767ee1dcf8c..6e27a4db6afec 100644
--- a/server/src/test/java/org/opensearch/indices/replication/OngoingSegmentReplicationsTests.java
+++ b/server/src/test/java/org/opensearch/indices/replication/OngoingSegmentReplicationsTests.java
@@ -17,6 +17,7 @@
import org.opensearch.common.util.CancellableThreads;
import org.opensearch.common.xcontent.XContentType;
import org.opensearch.index.IndexService;
+import org.opensearch.index.codec.CodecService;
import org.opensearch.index.engine.NRTReplicationEngineFactory;
import org.opensearch.index.shard.IndexShard;
import org.opensearch.index.shard.IndexShardTestCase;
@@ -47,7 +48,7 @@
public class OngoingSegmentReplicationsTests extends IndexShardTestCase {
private final IndicesService mockIndicesService = mock(IndicesService.class);
- private ReplicationCheckpoint testCheckpoint;
+ private ReplicationCheckpoint testCheckpoint, olderCodecTestCheckpoint;
private DiscoveryNode primaryDiscoveryNode;
private DiscoveryNode replicaDiscoveryNode;
private IndexShard primary;
@@ -73,8 +74,12 @@ public void setUp() throws Exception {
ShardId testShardId = primary.shardId();
+ CodecService codecService = new CodecService(null, null);
+ String defaultCodecName = codecService.codec(CodecService.DEFAULT_CODEC).getName();
+
// This mirrors the creation of the ReplicationCheckpoint inside CopyState
- testCheckpoint = new ReplicationCheckpoint(testShardId, primary.getOperationPrimaryTerm(), 0L, 0L);
+ testCheckpoint = new ReplicationCheckpoint(testShardId, primary.getOperationPrimaryTerm(), 0L, 0L, defaultCodecName);
+ olderCodecTestCheckpoint = new ReplicationCheckpoint(testShardId, primary.getOperationPrimaryTerm(), 0L, 0L, "Lucene94");
IndexService mockIndexService = mock(IndexService.class);
when(mockIndicesService.indexServiceSafe(testShardId.getIndex())).thenReturn(mockIndexService);
when(mockIndexService.getShard(testShardId.id())).thenReturn(primary);
@@ -89,6 +94,44 @@ public void tearDown() throws Exception {
super.tearDown();
}
+ public void testSuccessfulCodecCompatibilityCheck() throws Exception {
+ indexDoc(primary, "1", "{\"foo\" : \"baz\"}", XContentType.JSON, "foobar");
+ primary.refresh("Test");
+ OngoingSegmentReplications replications = spy(new OngoingSegmentReplications(mockIndicesService, recoverySettings));
+ // replica checkpoint is on same/higher lucene codec than primary
+ final CheckpointInfoRequest request = new CheckpointInfoRequest(
+ 1L,
+ replica.routingEntry().allocationId().getId(),
+ replicaDiscoveryNode,
+ testCheckpoint
+ );
+ final FileChunkWriter segmentSegmentFileChunkWriter = (fileMetadata, position, content, lastChunk, totalTranslogOps, listener) -> {
+ listener.onResponse(null);
+ };
+ final CopyState copyState = replications.prepareForReplication(request, segmentSegmentFileChunkWriter);
+ }
+
+ public void testFailCodecCompatibilityCheck() throws Exception {
+ indexDoc(primary, "1", "{\"foo\" : \"baz\"}", XContentType.JSON, "foobar");
+ primary.refresh("Test");
+ OngoingSegmentReplications replications = spy(new OngoingSegmentReplications(mockIndicesService, recoverySettings));
+ // replica checkpoint is on lower/older lucene codec than primary
+ final CheckpointInfoRequest request = new CheckpointInfoRequest(
+ 1L,
+ replica.routingEntry().allocationId().getId(),
+ replicaDiscoveryNode,
+ olderCodecTestCheckpoint
+ );
+ final FileChunkWriter segmentSegmentFileChunkWriter = (fileMetadata, position, content, lastChunk, totalTranslogOps, listener) -> {
+ listener.onResponse(null);
+ };
+ try {
+ final CopyState copyState = replications.prepareForReplication(request, segmentSegmentFileChunkWriter);
+ } catch (CancellableThreads.ExecutionCancelledException ex) {
+ Assert.assertTrue(ex.getMessage().contains("Requested unsupported codec version"));
+ }
+ }
+
public void testPrepareAndSendSegments() throws IOException {
indexDoc(primary, "1", "{\"foo\" : \"baz\"}", XContentType.JSON, "foobar");
primary.refresh("Test");
diff --git a/server/src/test/java/org/opensearch/indices/replication/PrimaryShardReplicationSourceTests.java b/server/src/test/java/org/opensearch/indices/replication/PrimaryShardReplicationSourceTests.java
index 10b747b822819..fdd707ae88195 100644
--- a/server/src/test/java/org/opensearch/indices/replication/PrimaryShardReplicationSourceTests.java
+++ b/server/src/test/java/org/opensearch/indices/replication/PrimaryShardReplicationSourceTests.java
@@ -8,6 +8,7 @@
package org.opensearch.indices.replication;
+import org.apache.lucene.codecs.Codec;
import org.apache.lucene.util.Version;
import org.junit.Assert;
import org.opensearch.action.ActionListener;
@@ -93,7 +94,13 @@ public void tearDown() throws Exception {
}
public void testGetCheckpointMetadata() {
- final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(indexShard.shardId(), PRIMARY_TERM, SEGMENTS_GEN, VERSION);
+ final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(
+ indexShard.shardId(),
+ PRIMARY_TERM,
+ SEGMENTS_GEN,
+ VERSION,
+ Codec.getDefault().getName()
+ );
replicationSource.getCheckpointMetadata(REPLICATION_ID, checkpoint, mock(ActionListener.class));
CapturingTransport.CapturedRequest[] requestList = transport.getCapturedRequestsAndClear();
assertEquals(1, requestList.length);
@@ -104,7 +111,13 @@ public void testGetCheckpointMetadata() {
}
public void testGetSegmentFiles() {
- final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(indexShard.shardId(), PRIMARY_TERM, SEGMENTS_GEN, VERSION);
+ final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(
+ indexShard.shardId(),
+ PRIMARY_TERM,
+ SEGMENTS_GEN,
+ VERSION,
+ Codec.getDefault().getName()
+ );
StoreFileMetadata testMetadata = new StoreFileMetadata("testFile", 1L, "checksum", Version.LATEST);
replicationSource.getSegmentFiles(
REPLICATION_ID,
@@ -126,7 +139,13 @@ public void testGetSegmentFiles() {
*/
public void testTransportTimeoutForGetSegmentFilesAction() {
long fileSize = (long) (Math.pow(10, 9));
- final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(indexShard.shardId(), PRIMARY_TERM, SEGMENTS_GEN, VERSION);
+ final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(
+ indexShard.shardId(),
+ PRIMARY_TERM,
+ SEGMENTS_GEN,
+ VERSION,
+ Codec.getDefault().getName()
+ );
StoreFileMetadata testMetadata = new StoreFileMetadata("testFile", fileSize, "checksum", Version.LATEST);
replicationSource.getSegmentFiles(
REPLICATION_ID,
@@ -145,7 +164,13 @@ public void testTransportTimeoutForGetSegmentFilesAction() {
public void testGetSegmentFiles_CancelWhileRequestOpen() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
- final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(indexShard.shardId(), PRIMARY_TERM, SEGMENTS_GEN, VERSION);
+ final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(
+ indexShard.shardId(),
+ PRIMARY_TERM,
+ SEGMENTS_GEN,
+ VERSION,
+ Codec.getDefault().getName()
+ );
StoreFileMetadata testMetadata = new StoreFileMetadata("testFile", 1L, "checksum", Version.LATEST);
replicationSource.getSegmentFiles(
REPLICATION_ID,
diff --git a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationSourceServiceTests.java b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationSourceServiceTests.java
index 0d05b1ec8679e..41022b77b46e1 100644
--- a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationSourceServiceTests.java
+++ b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationSourceServiceTests.java
@@ -8,6 +8,7 @@
package org.opensearch.indices.replication;
+import org.apache.lucene.codecs.Codec;
import org.opensearch.Version;
import org.opensearch.action.ActionListener;
import org.opensearch.cluster.node.DiscoveryNode;
@@ -55,7 +56,13 @@ public void setUp() throws Exception {
when(mockIndexService.getShard(testShardId.id())).thenReturn(mockIndexShard);
// This mirrors the creation of the ReplicationCheckpoint inside CopyState
- testCheckpoint = new ReplicationCheckpoint(testShardId, mockIndexShard.getOperationPrimaryTerm(), 0L, 0L);
+ testCheckpoint = new ReplicationCheckpoint(
+ testShardId,
+ mockIndexShard.getOperationPrimaryTerm(),
+ 0L,
+ 0L,
+ Codec.getDefault().getName()
+ );
testThreadPool = new TestThreadPool("test", Settings.EMPTY);
CapturingTransport transport = new CapturingTransport();
localNode = new DiscoveryNode("local", buildNewFakeTransportAddress(), Version.CURRENT);
diff --git a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java
index bae0afb5bcc3b..357a88c27fc46 100644
--- a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java
+++ b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java
@@ -15,6 +15,7 @@
import org.opensearch.cluster.metadata.IndexMetadata;
import org.opensearch.common.settings.Settings;
import org.opensearch.common.util.CancellableThreads;
+import org.opensearch.index.codec.CodecService;
import org.opensearch.index.engine.NRTReplicationEngineFactory;
import org.opensearch.index.replication.TestReplicationSource;
import org.opensearch.index.shard.IndexShard;
@@ -62,10 +63,12 @@ public void setUp() throws Exception {
.put(IndexMetadata.SETTING_REPLICATION_TYPE, ReplicationType.SEGMENT)
.put("node.name", SegmentReplicationTargetServiceTests.class.getSimpleName())
.build();
+ CodecService codecService = new CodecService(null, null);
+ String defaultCodecName = codecService.codec(CodecService.DEFAULT_CODEC).getName();
primaryShard = newStartedShard(true, settings);
replicaShard = newShard(false, settings, new NRTReplicationEngineFactory());
recoverReplica(replicaShard, primaryShard, true, getReplicationFunc(replicaShard));
- checkpoint = new ReplicationCheckpoint(replicaShard.shardId(), 0L, 0L, 0L);
+ checkpoint = new ReplicationCheckpoint(replicaShard.shardId(), 0L, 0L, 0L, defaultCodecName);
SegmentReplicationSourceFactory replicationSourceFactory = mock(SegmentReplicationSourceFactory.class);
replicationSource = mock(SegmentReplicationSource.class);
when(replicationSourceFactory.get(replicaShard)).thenReturn(replicationSource);
@@ -76,13 +79,15 @@ public void setUp() throws Exception {
initialCheckpoint.getShardId(),
initialCheckpoint.getPrimaryTerm(),
initialCheckpoint.getSegmentsGen(),
- initialCheckpoint.getSegmentInfosVersion() + 1
+ initialCheckpoint.getSegmentInfosVersion() + 1,
+ defaultCodecName
);
newPrimaryCheckpoint = new ReplicationCheckpoint(
initialCheckpoint.getShardId(),
initialCheckpoint.getPrimaryTerm() + 1,
initialCheckpoint.getSegmentsGen(),
- initialCheckpoint.getSegmentInfosVersion() + 1
+ initialCheckpoint.getSegmentInfosVersion() + 1,
+ defaultCodecName
);
}
diff --git a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetTests.java b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetTests.java
index 0c766c66413dd..a029d87f4a575 100644
--- a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetTests.java
+++ b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetTests.java
@@ -8,6 +8,7 @@
package org.opensearch.indices.replication;
+import org.apache.lucene.codecs.Codec;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.StringField;
@@ -106,7 +107,8 @@ public void setUp() throws Exception {
spyIndexShard.shardId(),
spyIndexShard.getPendingPrimaryTerm(),
testSegmentInfos.getGeneration(),
- testSegmentInfos.version
+ testSegmentInfos.version,
+ Codec.getDefault().getName()
);
}
diff --git a/server/src/test/java/org/opensearch/indices/replication/checkpoint/PublishCheckpointActionTests.java b/server/src/test/java/org/opensearch/indices/replication/checkpoint/PublishCheckpointActionTests.java
index 8a67292703da0..c851edf5e1bc8 100644
--- a/server/src/test/java/org/opensearch/indices/replication/checkpoint/PublishCheckpointActionTests.java
+++ b/server/src/test/java/org/opensearch/indices/replication/checkpoint/PublishCheckpointActionTests.java
@@ -8,6 +8,7 @@
package org.opensearch.indices.replication.checkpoint;
+import org.apache.lucene.codecs.Codec;
import org.opensearch.action.ActionListener;
import org.opensearch.action.support.ActionFilters;
import org.opensearch.action.support.ActionTestUtils;
@@ -104,7 +105,7 @@ public void testPublishCheckpointActionOnPrimary() {
mockTargetService
);
- final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(indexShard.shardId(), 1111, 11, 1);
+ final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(indexShard.shardId(), 1111, 11, 1, Codec.getDefault().getName());
final PublishCheckpointRequest request = new PublishCheckpointRequest(checkpoint);
action.shardOperationOnPrimary(request, indexShard, ActionTestUtils.assertNoFailureListener(result -> {
@@ -139,7 +140,7 @@ public void testPublishCheckpointActionOnReplica() {
mockTargetService
);
- final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(indexShard.shardId(), 1111, 11, 1);
+ final ReplicationCheckpoint checkpoint = new ReplicationCheckpoint(indexShard.shardId(), 1111, 11, 1, Codec.getDefault().getName());
final PublishCheckpointRequest request = new PublishCheckpointRequest(checkpoint);
diff --git a/server/src/test/java/org/opensearch/indices/replication/common/CopyStateTests.java b/server/src/test/java/org/opensearch/indices/replication/common/CopyStateTests.java
index a87a8de206a39..e3b48302ae6ef 100644
--- a/server/src/test/java/org/opensearch/indices/replication/common/CopyStateTests.java
+++ b/server/src/test/java/org/opensearch/indices/replication/common/CopyStateTests.java
@@ -8,12 +8,14 @@
package org.opensearch.indices.replication.common;
+import org.apache.lucene.codecs.Codec;
import org.apache.lucene.index.IndexCommit;
import org.apache.lucene.index.IndexFileNames;
import org.apache.lucene.index.SegmentInfos;
import org.apache.lucene.util.Version;
import org.opensearch.common.collect.Tuple;
import org.opensearch.common.concurrent.GatedCloseable;
+import org.opensearch.index.codec.CodecService;
import org.opensearch.index.shard.IndexShard;
import org.opensearch.index.shard.IndexShardTestCase;
import org.opensearch.index.shard.ShardId;
@@ -49,7 +51,10 @@ public class CopyStateTests extends IndexShardTestCase {
public void testCopyStateCreation() throws IOException {
final IndexShard mockIndexShard = createMockIndexShard();
- CopyState copyState = new CopyState(ReplicationCheckpoint.empty(mockIndexShard.shardId()), mockIndexShard);
+ CopyState copyState = new CopyState(
+ ReplicationCheckpoint.empty(mockIndexShard.shardId(), new CodecService(null, null).codec("default").getName()),
+ mockIndexShard
+ );
ReplicationCheckpoint checkpoint = copyState.getCheckpoint();
assertEquals(TEST_SHARD_ID, checkpoint.getShardId());
// version was never set so this should be zero
@@ -67,7 +72,13 @@ public static IndexShard createMockIndexShard() throws IOException {
when(mockShard.store()).thenReturn(mockStore);
SegmentInfos testSegmentInfos = new SegmentInfos(Version.LATEST.major);
- ReplicationCheckpoint testCheckpoint = new ReplicationCheckpoint(mockShard.shardId(), mockShard.getOperationPrimaryTerm(), 0L, 0L);
+ ReplicationCheckpoint testCheckpoint = new ReplicationCheckpoint(
+ mockShard.shardId(),
+ mockShard.getOperationPrimaryTerm(),
+ 0L,
+ 0L,
+ Codec.getDefault().getName()
+ );
final Tuple, ReplicationCheckpoint> gatedCloseableReplicationCheckpointTuple = new Tuple<>(
new GatedCloseable<>(testSegmentInfos, () -> {}),
testCheckpoint
diff --git a/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java b/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java
index 7b81bd45cc8b6..80e603bd80420 100644
--- a/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java
+++ b/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java
@@ -1299,7 +1299,10 @@ public void getCheckpointMetadata(
ActionListener listener
) {
try {
- final CopyState copyState = new CopyState(ReplicationCheckpoint.empty(primaryShard.shardId), primaryShard);
+ final CopyState copyState = new CopyState(
+ ReplicationCheckpoint.empty(primaryShard.shardId, primaryShard.getDefaultCodecName()),
+ primaryShard
+ );
listener.onResponse(
new CheckpointInfoResponse(copyState.getCheckpoint(), copyState.getMetadataMap(), copyState.getInfosBytes())
);
@@ -1353,7 +1356,7 @@ public final List replicateSegments(IndexShard primary
for (IndexShard replica : replicaShards) {
final SegmentReplicationTargetService targetService = prepareForReplication(primaryShard, replica);
final SegmentReplicationTarget target = targetService.startReplication(
- ReplicationCheckpoint.empty(replica.shardId),
+ ReplicationCheckpoint.empty(replica.shardId, replica.getDefaultCodecName()),
replica,
new SegmentReplicationTargetService.SegmentReplicationListener() {
@Override