-
Notifications
You must be signed in to change notification settings - Fork 1.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactor Translog metadata upload/download to write/read header and f…
…ooter via VersionedCodecStreamWrapper (#7953) Signed-off-by: Bhumika Saini <[email protected]> (cherry picked from commit 4b4d84e) Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
- Loading branch information
1 parent
c83c683
commit 5d72f05
Showing
5 changed files
with
201 additions
and
54 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
63 changes: 63 additions & 0 deletions
63
...src/main/java/org/opensearch/index/translog/transfer/TranslogTransferMetadataHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| /* | ||
| * 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.translog.transfer; | ||
|
|
||
| import org.apache.lucene.store.IndexInput; | ||
| import org.apache.lucene.store.IndexOutput; | ||
| import org.opensearch.common.io.IndexIOStreamHandler; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| /** | ||
| * Handler for {@link TranslogTransferMetadata} | ||
| * | ||
| * @opensearch.internal | ||
| */ | ||
| public class TranslogTransferMetadataHandler implements IndexIOStreamHandler<TranslogTransferMetadata> { | ||
|
|
||
| /** | ||
| * Implements logic to read content from file input stream {@code indexInput} and parse into {@link TranslogTransferMetadata} | ||
| * | ||
| * @param indexInput file input stream | ||
| * @return content parsed to {@link TranslogTransferMetadata} | ||
| */ | ||
| @Override | ||
| public TranslogTransferMetadata readContent(IndexInput indexInput) throws IOException { | ||
| long primaryTerm = indexInput.readLong(); | ||
| long generation = indexInput.readLong(); | ||
| long minTranslogGeneration = indexInput.readLong(); | ||
| Map<String, String> generationToPrimaryTermMapper = indexInput.readMapOfStrings(); | ||
|
|
||
| int count = generationToPrimaryTermMapper.size(); | ||
| TranslogTransferMetadata metadata = new TranslogTransferMetadata(primaryTerm, generation, minTranslogGeneration, count); | ||
| metadata.setGenerationToPrimaryTermMapper(generationToPrimaryTermMapper); | ||
|
|
||
| return metadata; | ||
| } | ||
|
|
||
| /** | ||
| * Implements logic to write content from {@code content} to file output stream {@code indexOutput} | ||
| * | ||
| * @param indexOutput file input stream | ||
| * @param content metadata content to be written | ||
| */ | ||
| @Override | ||
| public void writeContent(IndexOutput indexOutput, TranslogTransferMetadata content) throws IOException { | ||
| indexOutput.writeLong(content.getPrimaryTerm()); | ||
| indexOutput.writeLong(content.getGeneration()); | ||
| indexOutput.writeLong(content.getMinTranslogGeneration()); | ||
| if (content.getGenerationToPrimaryTermMapper() != null) { | ||
| indexOutput.writeMapOfStrings(content.getGenerationToPrimaryTermMapper()); | ||
| } else { | ||
| indexOutput.writeMapOfStrings(new HashMap<>()); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
93 changes: 93 additions & 0 deletions
93
...est/java/org/opensearch/index/translog/transfer/TranslogTransferMetadataHandlerTests.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| /* | ||
| * 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.translog.transfer; | ||
|
|
||
| import org.apache.lucene.store.IndexInput; | ||
| import org.apache.lucene.store.OutputStreamIndexOutput; | ||
| import org.junit.Before; | ||
| import org.opensearch.common.bytes.BytesReference; | ||
| import org.opensearch.common.io.stream.BytesStreamOutput; | ||
| import org.opensearch.common.lucene.store.ByteArrayIndexInput; | ||
| import org.opensearch.test.OpenSearchTestCase; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| public class TranslogTransferMetadataHandlerTests extends OpenSearchTestCase { | ||
| private TranslogTransferMetadataHandler handler; | ||
|
|
||
| @Before | ||
| public void setUp() throws Exception { | ||
| super.setUp(); | ||
| handler = new TranslogTransferMetadataHandler(); | ||
| } | ||
|
|
||
| public void testReadContent() throws IOException { | ||
| TranslogTransferMetadata expectedMetadata = getTestMetadata(); | ||
|
|
||
| // Operation: Read expected metadata from source input stream. | ||
| IndexInput indexInput = new ByteArrayIndexInput("metadata file", getTestMetadataBytes()); | ||
| TranslogTransferMetadata actualMetadata = handler.readContent(indexInput); | ||
|
|
||
| // Verification: Compare actual metadata read from the source input stream. | ||
| assertEquals(expectedMetadata, actualMetadata); | ||
| } | ||
|
|
||
| public void testWriteContent() throws IOException { | ||
| TranslogTransferMetadata expectedMetadata = getTestMetadata(); | ||
|
|
||
| // Operation: Write expected metadata to the target output stream. | ||
| BytesStreamOutput output = new BytesStreamOutput(); | ||
| OutputStreamIndexOutput actualMetadataStream = new OutputStreamIndexOutput("dummy bytes", "dummy stream", output, 4096); | ||
| handler.writeContent(actualMetadataStream, expectedMetadata); | ||
| actualMetadataStream.close(); | ||
|
|
||
| // Verification: Compare actual metadata written to the target output stream. | ||
| IndexInput indexInput = new ByteArrayIndexInput("metadata file", BytesReference.toBytes(output.bytes())); | ||
| long primaryTerm = indexInput.readLong(); | ||
| long generation = indexInput.readLong(); | ||
| long minTranslogGeneration = indexInput.readLong(); | ||
| Map<String, String> generationToPrimaryTermMapper = indexInput.readMapOfStrings(); | ||
| int count = generationToPrimaryTermMapper.size(); | ||
| TranslogTransferMetadata actualMetadata = new TranslogTransferMetadata(primaryTerm, generation, minTranslogGeneration, count); | ||
| actualMetadata.setGenerationToPrimaryTermMapper(generationToPrimaryTermMapper); | ||
| assertEquals(expectedMetadata, actualMetadata); | ||
| } | ||
|
|
||
| private TranslogTransferMetadata getTestMetadata() { | ||
| long primaryTerm = 3; | ||
| long generation = 500; | ||
| long minTranslogGeneration = 300; | ||
| Map<String, String> generationToPrimaryTermMapper = new HashMap<>(); | ||
| generationToPrimaryTermMapper.put("300", "1"); | ||
| generationToPrimaryTermMapper.put("400", "2"); | ||
| generationToPrimaryTermMapper.put("500", "3"); | ||
| int count = generationToPrimaryTermMapper.size(); | ||
| TranslogTransferMetadata metadata = new TranslogTransferMetadata(primaryTerm, generation, minTranslogGeneration, count); | ||
| metadata.setGenerationToPrimaryTermMapper(generationToPrimaryTermMapper); | ||
|
|
||
| return metadata; | ||
| } | ||
|
|
||
| private byte[] getTestMetadataBytes() throws IOException { | ||
| TranslogTransferMetadata metadata = getTestMetadata(); | ||
|
|
||
| BytesStreamOutput output = new BytesStreamOutput(); | ||
| OutputStreamIndexOutput indexOutput = new OutputStreamIndexOutput("dummy bytes", "dummy stream", output, 4096); | ||
| indexOutput.writeLong(metadata.getPrimaryTerm()); | ||
| indexOutput.writeLong(metadata.getGeneration()); | ||
| indexOutput.writeLong(metadata.getMinTranslogGeneration()); | ||
| Map<String, String> generationToPrimaryTermMapper = metadata.getGenerationToPrimaryTermMapper(); | ||
| indexOutput.writeMapOfStrings(generationToPrimaryTermMapper); | ||
| indexOutput.close(); | ||
|
|
||
| return BytesReference.toBytes(output.bytes()); | ||
| } | ||
| } |