Skip to content

Commit

Permalink
IGNITE-22020 Add 'keepRaw' parameter to DumpReaderConfiguration (#11309)
Browse files Browse the repository at this point in the history
  • Loading branch information
@Vladsz83
Vladsz83 committed Apr 12, 2024
1 parent 327df73 commit 1fb6ad0
Show file tree
Hide file tree
Showing 4 changed files with 122 additions and 5 deletions.
2 changes: 1 addition & 1 deletion modules/core/src/main/java/org/apache/ignite/dump/DumpReader.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ public DumpReader(DumpReaderConfiguration cfg, IgniteLogger log) {
@Override public void run() {
ackAsciiLogo();

try (Dump dump = new Dump(cfg.dumpRoot(), null, cfg.keepBinary(), false, encryptionSpi(), log)) {
try (Dump dump = new Dump(cfg.dumpRoot(), null, cfg.keepBinary(), cfg.keepRaw(), encryptionSpi(), log)) {
DumpConsumer cnsmr = cfg.consumer();

cnsmr.start();
Expand Down
36 changes: 32 additions & 4 deletions modules/core/src/main/java/org/apache/ignite/dump/DumpReaderConfiguration.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import java.io.File;
import java.time.Duration;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.lang.IgniteExperimental;
Expand Down Expand Up @@ -53,9 +54,18 @@ public class DumpReaderConfiguration {
/** Stop processing partitions if consumer fail to process one. */
private final boolean failFast;

/** If {@code true} then don't deserialize {@link KeyCacheObject} and {@link CacheObject}. */
/**
* If {@code true} and if {@link #keepRaw} is {@code false} then keeps {@link DumpEntry#key()} and
* {@link DumpEntry#value()} as {@link BinaryObject}.
*/
private final boolean keepBinary;

/**
* If {@code true}, doesn't deserialize cache data and keeps {@link DumpEntry#key()} as {@link KeyCacheObject} and
* {@link DumpEntry#value()} as {@link CacheObject}. If {@code true}, disables {@link #keepBinary}.
*/
private final boolean keepRaw;

/** Cache group names. */
private final String[] cacheGrpNames;

Expand All @@ -70,7 +80,7 @@ public class DumpReaderConfiguration {
* @param cnsmr Dump consumer.
*/
public DumpReaderConfiguration(File dir, DumpConsumer cnsmr) {
this(dir, cnsmr, DFLT_THREAD_CNT, DFLT_TIMEOUT, true, true, null, false, null);
this(dir, cnsmr, DFLT_THREAD_CNT, DFLT_TIMEOUT, true, true, false, null, false, null);
}

/**
Expand All @@ -79,7 +89,11 @@ public DumpReaderConfiguration(File dir, DumpConsumer cnsmr) {
* @param thCnt Count of threads to consume dumped partitions.
* @param timeout Timeout of dump reader invocation.
* @param failFast Stop processing partitions if consumer fail to process one.
* @param keepBinary If {@code true} then don't deserialize {@link KeyCacheObject} and {@link CacheObject}.
* @param keepBinary If {@code true} and if {@link #keepRaw} is {@code false} then keeps {@link DumpEntry#key()} and
* {@link DumpEntry#value()} as {@link BinaryObject}.
* @param keepRaw If {@code true}, doesn't deserialize cache data and keeps {@link DumpEntry#key()} as
* {@link KeyCacheObject} and {@link DumpEntry#value()} as {@link CacheObject}. If {@code true},
* disables {@link #keepBinary}.
* @param cacheGrpNames Cache group names.
* @param skipCopies Skip copies.
* @param encSpi Encryption SPI.
Expand All @@ -91,6 +105,7 @@ public DumpReaderConfiguration(
Duration timeout,
boolean failFast,
boolean keepBinary,
boolean keepRaw,
String[] cacheGrpNames,
boolean skipCopies,
EncryptionSpi encSpi
Expand All @@ -101,6 +116,7 @@ public DumpReaderConfiguration(
this.timeout = timeout;
this.failFast = failFast;
this.keepBinary = keepBinary;
this.keepRaw = keepRaw;
this.cacheGrpNames = cacheGrpNames;
this.skipCopies = skipCopies;
this.encSpi = encSpi;
Expand Down Expand Up @@ -131,11 +147,23 @@ public boolean failFast() {
return failFast;
}

/** @return If {@code true} then don't deserialize {@link KeyCacheObject} and {@link CacheObject}. */
/**
* Actual only if {@link #keepRaw} is {@code false}.
*
* @return {@code True} if {@link DumpEntry#key()} and {@link DumpEntry#value()} are kept as {@link BinaryObject}.
*/
public boolean keepBinary() {
return keepBinary;
}

/**
* @return {@code True} if {@link DumpEntry#key()} and {@link DumpEntry#value()} are kept as {@link KeyCacheObject}
* and {@link CacheObject} respectively.
*/
public boolean keepRaw() {
return keepRaw;
}

/** @return Cache group names. */
public String[] cacheGroupNames() {
return cacheGrpNames;
Expand Down
1 change: 1 addition & 0 deletions ...che/ignite/internal/processors/cache/persistence/snapshot/dump/AbstractCacheDumpTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -428,6 +428,7 @@ else if (!cacheName.equals(DEFAULT_CACHE_NAME))
DFLT_THREAD_CNT, DFLT_TIMEOUT,
true,
false,
false,
cacheGrpNames,
skipCopies,
null
Expand Down
88 changes: 88 additions & 0 deletions .../ignite/internal/processors/cache/persistence/snapshot/dump/IgniteCacheDumpSelf2Test.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,16 +34,19 @@
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.zip.ZipInputStream;
import org.apache.commons.io.IOUtils;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteException;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheEntryVersion;
Expand Down Expand Up @@ -83,6 +86,7 @@
import org.apache.ignite.internal.processors.cacheobject.UserCacheObjectImpl;
import org.apache.ignite.internal.processors.dr.GridDrType;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFuture;
Expand Down Expand Up @@ -170,6 +174,86 @@ public class IgniteCacheDumpSelf2Test extends GridCommonAbstractTest {
cleanPersistenceDir();
}

/** */
@Test
public void testDumpRawData() throws Exception {
IgniteEx ign = startGrids(3);

Ignite cli = startClientGrid(G.allGrids().size());

cli.createCache(defaultCacheConfiguration());

for (int i = 0; i < KEYS_CNT; ++i)
cli.cache(DEFAULT_CACHE_NAME).put(i, USER_FACTORY.apply(i));

cli.snapshot().createDump(DMP_NAME, null).get();

AtomicBoolean keepRaw = new AtomicBoolean();
AtomicBoolean keepBinary = new AtomicBoolean();

DumpConsumer cnsmr = new DumpConsumer() {
@Override public void start() {
// No-op.
}

@Override public void onMappings(Iterator<TypeMapping> mappings) {
// No-op.
}

@Override public void onTypes(Iterator<BinaryType> types) {
// No-op.
}

@Override public void onCacheConfigs(Iterator<StoredCacheData> caches) {
// No-op.
}

@Override public void onPartition(int grp, int part, Iterator<DumpEntry> data) {
data.forEachRemaining(de -> {
if (keepRaw.get()) {
assertTrue(de.key() instanceof KeyCacheObject);
assertTrue(de.value() instanceof CacheObject);
}
else {
assertTrue(de.key() instanceof Integer);

if (keepBinary.get())
assertTrue(de.value() instanceof BinaryObject);
else
assertTrue(de.value() instanceof User);
}
});
}

@Override public void stop() {
// No-op.
}
};

for (boolean raw : Arrays.asList(true, false)) {
for (boolean binary : Arrays.asList(true, false)) {
keepRaw.set(raw);
keepBinary.set(binary);

new DumpReader(
new DumpReaderConfiguration(
dumpDirectory(ign, DMP_NAME),
cnsmr,
DFLT_THREAD_CNT,
DFLT_TIMEOUT,
true,
keepBinary.get(),
keepRaw.get(),
null,
false,
null
),
log
).run();
}
}
}

/** Checks a dump when it is created with the data streamer just after a restart. */
@Test
public void testDumpAfterRestartWithStreamer() throws Exception {
Expand Down Expand Up @@ -748,6 +832,7 @@ public void testDumpEntryConflictVersion() throws Exception {
DFLT_TIMEOUT,
true,
false,
false,
null,
false,
null
Expand Down Expand Up @@ -812,6 +897,7 @@ public void testReadEncrypted() throws Exception {
DFLT_TIMEOUT,
true,
false,
false,
null,
false,
null
Expand Down Expand Up @@ -840,6 +926,7 @@ public void testReadEncrypted() throws Exception {
DFLT_TIMEOUT,
true,
false,
false,
null,
false,
encSpi
Expand Down Expand Up @@ -871,6 +958,7 @@ public void testReadEncrypted() throws Exception {
DFLT_TIMEOUT,
true,
false,
false,
null,
false,
encryptionSpi()
Expand Down

0 comments on commit 1fb6ad0

Please sign in to comment.