diff --git a/.gitmodules b/.gitmodules index 616dacf610a7..38d6bc3bbc62 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,4 +1,4 @@ [submodule "modules/accord"] path = modules/accord - url = https://github.com/apache/cassandra-accord.git - branch = trunk + url = https://github.com/alanwang67/cassandra-accord.git + branch = accordJournalDeserializeNextFix diff --git a/modules/accord b/modules/accord index c9ec676c4cd3..ee2f4d2c0100 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit c9ec676c4cd3adced869301a38714c827ac9789e +Subproject commit ee2f4d2c010053b45fcbb844bbc8cb39e34d233e diff --git a/src/java/org/apache/cassandra/service/accord/journal/CommandChanges.java b/src/java/org/apache/cassandra/service/accord/journal/CommandChanges.java index 13c88f283490..50a639283709 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/CommandChanges.java +++ b/src/java/org/apache/cassandra/service/accord/journal/CommandChanges.java @@ -238,8 +238,8 @@ public void deserializeNext(DataInputPlus in, Version userVersion) throws IOExce for (CommandChange.Field field = nextSetField(iterable); field != null; field = nextSetField(iterable = unsetIterable(field, iterable))) { // Since we are iterating in reverse order, we skip the fields that were - // set by entries written later (i.e. already read ones). - if (isChanged(field, flags | mask) && field != CLEANUP) + // set by entries written later (i.e. already read ones) or if the mask did not include the field. + if ((isChanged(field, flags) || ((mask & (1 << field.ordinal())) != 0)) && field != CLEANUP) skip(txnId, field, in, userVersion); else deserialize(field, in, userVersion); diff --git a/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java b/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java index cc212ea9495d..6f1c50b48d9d 100644 --- a/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java +++ b/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java @@ -29,6 +29,7 @@ import accord.impl.CommandChange; import accord.local.Command; import accord.local.RedundantBefore; +import accord.primitives.Ballot; import accord.primitives.SaveStatus; import accord.primitives.Status; import accord.primitives.TxnId; @@ -102,36 +103,64 @@ public void serde() Gen gen = AccordGenerators.commandsBuilder(); try (DataOutputBuffer out = new DataOutputBuffer()) { - qt().forAll(gen) - .check(cmdBuilder -> { - for (Version version : Version.V1.greaterThanOrEqual()) + qt().check(rs -> { + AccordGenerators.CommandBuilder cmdBuilder = gen.next(rs); + for (Version version : Version.V1.greaterThanOrEqual()) + { + SoftAssertions checks = new SoftAssertions(); + for (SaveStatus saveStatus : SaveStatus.values()) { - SoftAssertions checks = new SoftAssertions(); - for (SaveStatus saveStatus : SaveStatus.values()) + if (cmdBuilder.txnId.awaitsOnlyDeps() && saveStatus.is(Status.Truncated)) + continue; + + out.clear(); + Command orig = cmdBuilder.build(saveStatus); + CommandChangeWriter writer = CommandChangeWriter.make(null, orig); + if (writer == null) + continue; + + writer.write(out, version); + Load load = Load.values()[rs.nextInt(Load.values().length)]; + CommandChanges builder = new CommandChanges(orig.txnId(), load); + builder.deserializeNext(new DataInputBuffer(out.unsafeGetBufferAndFlip(), false), version); + + if (load != Load.ALL) { - if (cmdBuilder.txnId.awaitsOnlyDeps() && saveStatus.is(Status.Truncated)) - continue; - - out.clear(); - Command orig = cmdBuilder.build(saveStatus); - CommandChangeWriter writer = CommandChangeWriter.make(null, orig); - if (writer == null) - continue; - - writer.write(out, version); - CommandChanges builder = new CommandChanges(orig.txnId(), Load.ALL); - builder.deserializeNext(new DataInputBuffer(out.unsafeGetBufferAndFlip(), false), version); - // We are not persisting the result, so force it for strict equality - builder.forceResult(orig.result()); - - Command reconstructed = builder.construct(RedundantBefore.EMPTY); - - checks.assertThat(reconstructed) - .describedAs("lhs=expected\nrhs=actual\n%s", new LazyToString(() -> ReflectionUtils.recursiveEquals(orig, reconstructed).toString())) - .isEqualTo(orig); + if (!CommandChange.isNull(Field.SAVE_STATUS, getFlags(null, orig))) + checks.assertThat(builder.saveStatus()).isEqualTo(orig.saveStatus()); + if (!CommandChange.isNull(Field.PARTICIPANTS, getFlags(null, orig))) + checks.assertThat(builder.participants()).isEqualTo(orig.participants()); + if (!CommandChange.isNull(Field.EXECUTE_AT, getFlags(null, orig))) + checks.assertThat(builder.executeAt()).isEqualTo(orig.executeAt()); + if (!CommandChange.isNull(Field.DURABILITY, getFlags(null, orig))) + checks.assertThat(builder.durability()).isEqualTo(orig.durability()); + if (load == Load.MINIMAL_WITH_DEPS && !CommandChange.isNull(Field.PARTIAL_DEPS, getFlags(null, orig))) + checks.assertThat(builder.partialDeps()).isEqualTo(orig.partialDeps()); + int mask = CommandChange.mask(load); + + // Ensure that fields that are masked out are equal to their default values + for (Field field : ALL) + { + if (field == Field.CLEANUP || (mask & (1 << field.ordinal())) == 0) + continue; + Object unset = field == Field.PROMISED || field == Field.ACCEPTED ? Ballot.ZERO + : field == Field.MIN_UNIQUE_HLC ? 0L : null; + checks.assertThat(builder.get(field)).isEqualTo(unset); + } + continue; } - checks.assertAll(); + + // We are not persisting the result, so force it for strict equality + builder.forceResult(orig.result()); + + Command reconstructed = builder.construct(RedundantBefore.EMPTY); + + checks.assertThat(reconstructed) + .describedAs("lhs=expected\nrhs=actual\n%s", new LazyToString(() -> ReflectionUtils.recursiveEquals(orig, reconstructed).toString())) + .isEqualTo(orig); } + checks.assertAll(); + } }); } }