Skip to content

Commit 820a3c0

Browse files
szymon-miezaldriftx
szymon-miezal
authored and
driftx
committed
HCD-241: DSE 6.8/9 upgradeability (#2240)
This PR addresses critical compatibility issues discovered during DSE to HCD (Hyper-Converged Database) upgrade scenarios. The changes focus on ensuring smooth interoperability between DSE 6.8+ nodes and HCD nodes during mixed-version cluster operations. **Problem:** When upgrading from DSE to HCD, troubleshooting connection and gossip issues was difficult due to insufficient logging. **Solution:** Added comprehensive logging throughout the handshake and gossip processes, including: - Port selection logic for DSE → HCD upgrades - Messaging version negotiation during handshake - Protocol proposals from DSE peers - Gossip state values (both loaded and saved) - Connection initialization events **Why it matters:** This gives operators visibility into what's happening during the upgrade process, making it much easier to diagnose and resolve issues in production environments. --- **Problem:** Nodes would crash with `ArrayIndexOutOfBoundsException` when processing gossip messages from pre-4.0 nodes. This happened because the code assumed certain delimiters would always be present in address and status values, but older nodes sent data in different formats. Both INTERNAL_ADDRESS_AND_PORT and NATIVE_ADDRESS_AND_PORT could arrive without port delimiters (e.g., "10.0.0.1" instead of "10.0.0.1:7000"), and STATUS_WITH_PORT could arrive without additional port information. **Solution:** Added defensive length checks after splitting gossip values to gracefully handle cases where expected delimiters are missing: - Internal and native IP addresses without ports (e.g., "10.0.0.1" vs "10.0.0.1:7000") - Status values without additional port information **Why it matters:** Prevents cluster instability and crashes during mixed-version operations. --- **Problem:** During startup connectivity checks, HCD nodes were attempting to send PING requests to DSE 6.8+ peers, but these versions don't support PING_REQ messages (similar to DSE 6.x and Cassandra 3.x). This resulted in noisy error logs during cluster startup. **Solution:** Extended the existing version detection logic to recognize and skip PING requests for DSE 6.8+ peers during the startup connectivity check phase. **Why it matters:** Eliminates unnecessary error logs when HCD nodes join or restart in a mixed DSE/HCD environment. --- **Problem:** When keyspace schemas were updated during migration, User-Defined Types (UDTs) from the previous schema could be lost if they weren't explicitly present in the new schema. This caused failures when inherited tables depended on these types. **Solution:** Modified the schema transformation logic to preserve UDTs from the previous schema when they don't exist in the new schema, ensuring dependent tables continue to function correctly. **Why it matters:** Prevents data model corruption and application failures during schema migrations, particularly important when dealing with complex schemas that use inheritance and UDTs. --- These changes collectively improve the robustness and reliability of DSE to HCD upgrades by: - Making issues easier to diagnose through better logging - Preventing crashes from malformed gossip data - Ensuring protocol compatibility across versions - Preserving schema integrity during migrations
1 parent 7deb016 commit 820a3c0

7 files changed

Lines changed: 475 additions & 7 deletions

File tree

‎src/java/org/apache/cassandra/gms/EndpointState.java‎

Lines changed: 108 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -484,20 +484,21 @@ static VersionedValue serializerFilterValue(ApplicationState state, VersionedVal
484484

485485
class EndpointStateSerializer implements IVersionedSerializer<EndpointState>
486486
{
487+
private static final Logger logger = LoggerFactory.getLogger(EndpointStateSerializer.class);
488+
487489
public void serialize(EndpointState epState, DataOutputPlus out, int version) throws IOException
488490
{
489491
/* serialize the HeartBeatState */
490492
HeartBeatState hbState = epState.getHeartBeatState();
491493
HeartBeatState.serializer.serialize(hbState, out, version);
492494

493495
/* serialize the map of ApplicationState objects */
494-
Set<Map.Entry<ApplicationState, VersionedValue>> states = epState.states();
496+
Set<Map.Entry<ApplicationState, VersionedValue>> states = filterOutgoingStates(epState.states(), version);
495497
out.writeInt(states.size());
496498
for (Map.Entry<ApplicationState, VersionedValue> state : states)
497499
{
498-
VersionedValue value = filterValue(state.getKey(), state.getValue(), version);
499500
out.writeInt(state.getKey().ordinal());
500-
VersionedValue.serializer.serialize(value, out, version);
501+
VersionedValue.serializer.serialize(state.getValue(), out, version);
501502
}
502503
}
503504

@@ -514,19 +515,18 @@ public EndpointState deserialize(DataInputPlus in, int version) throws IOExcepti
514515
states.put(Gossiper.STATES[key], value);
515516
}
516517

517-
return new EndpointState(hbState, states);
518+
return new EndpointState(hbState, filterIncomingStates(states, version));
518519
}
519520

520521
public long serializedSize(EndpointState epState, int version)
521522
{
522523
long size = HeartBeatState.serializer.serializedSize(epState.getHeartBeatState(), version);
523-
Set<Map.Entry<ApplicationState, VersionedValue>> states = epState.states();
524+
Set<Map.Entry<ApplicationState, VersionedValue>> states = filterOutgoingStates(epState.states(), version);
524525
size += TypeSizes.sizeof(states.size());
525526
for (Map.Entry<ApplicationState, VersionedValue> state : states)
526527
{
527-
VersionedValue value = filterValue(state.getKey(), state.getValue(), version);
528528
size += TypeSizes.sizeof(state.getKey().ordinal());
529-
size += VersionedValue.serializer.serializedSize(value, version);
529+
size += VersionedValue.serializer.serializedSize(state.getValue(), version);
530530
}
531531
return size;
532532
}
@@ -539,4 +539,105 @@ static VersionedValue filterValue(ApplicationState state, VersionedValue value,
539539
? VersionedValue.unsafeMakeVersionedValue(value.value.replaceFirst("-[0-9a-f]{7,40}$", ""), value.version)
540540
: value;
541541
}
542+
543+
@VisibleForTesting
544+
static Set<Map.Entry<ApplicationState, VersionedValue>> filterOutgoingStates(Set<Map.Entry<ApplicationState, VersionedValue>> states, int version)
545+
{
546+
if (version < MessagingService.VERSION_40)
547+
{
548+
Set<Map.Entry<ApplicationState, VersionedValue>> filteredStates = new HashSet<>();
549+
for (Map.Entry<ApplicationState, VersionedValue> state : states)
550+
filteredStates.addAll(filterOutgoingState(state, version).entrySet());
551+
return filteredStates;
552+
}
553+
return states;
554+
}
555+
556+
private static Map<ApplicationState, VersionedValue> filterOutgoingState(Map.Entry<ApplicationState, VersionedValue> state, int version)
557+
{
558+
assert version < MessagingService.VERSION_40;
559+
VersionedValue vv = state.getValue();
560+
if (logger.isTraceEnabled())
561+
logger.trace("Fetching the key from state {}({}) with value of {}", state.getKey(), state.getKey().ordinal(), vv);
562+
String[] values;
563+
switch (state.getKey())
564+
{
565+
case INTERNAL_ADDRESS_AND_PORT:
566+
values = splitAddressAndPort(vv);
567+
if (values.length > 1)
568+
return Map.of(ApplicationState.values()[7], VersionedValue.unsafeMakeVersionedValue(values[0], vv.version),
569+
ApplicationState.values()[17], VersionedValue.unsafeMakeVersionedValue(values[1], vv.version));
570+
else
571+
return Map.of(ApplicationState.values()[17], VersionedValue.unsafeMakeVersionedValue(vv.value, vv.version));
572+
case NATIVE_ADDRESS_AND_PORT:
573+
values = splitAddressAndPort(vv);
574+
if (values.length > 1)
575+
return Map.of(ApplicationState.values()[15], VersionedValue.unsafeMakeVersionedValue(values[1], vv.version));
576+
else
577+
return Map.of(ApplicationState.values()[15], VersionedValue.unsafeMakeVersionedValue(vv.value, vv.version));
578+
case STATUS_WITH_PORT:
579+
values = splitStatusAndPort(vv);
580+
if (values.length > 1)
581+
return Map.of(ApplicationState.values()[0], VersionedValue.unsafeMakeVersionedValue(values[0], vv.version));
582+
else
583+
return Map.of(ApplicationState.values()[0], VersionedValue.unsafeMakeVersionedValue(vv.value, vv.version));
584+
case DISK_USAGE:
585+
return Map.of(ApplicationState.values()[21], state.getValue());
586+
case SSTABLE_VERSIONS:
587+
return Map.of();
588+
case INDEX_STATUS:
589+
return Map.of();
590+
case RELEASE_VERSION:
591+
return Map.of(ApplicationState.RELEASE_VERSION, filterValue(state.getKey(), vv, version));
592+
default:
593+
return Map.of(state.getKey(), state.getValue());
594+
}
595+
}
596+
597+
private static String[] splitStatusAndPort(VersionedValue vv)
598+
{
599+
return vv.value.split("[:,]");
600+
}
601+
602+
private static String[] splitAddressAndPort(VersionedValue vv)
603+
{
604+
return vv.value.split(":");
605+
}
606+
607+
@VisibleForTesting
608+
static Map<ApplicationState, VersionedValue> filterIncomingStates(Map<ApplicationState, VersionedValue> states, int version)
609+
{
610+
if (version < MessagingService.VERSION_40)
611+
{
612+
Map<ApplicationState, VersionedValue> filteredStates = new EnumMap<>(ApplicationState.class);
613+
for (Map.Entry<ApplicationState, VersionedValue> state : states.entrySet())
614+
{
615+
VersionedValue vv = state.getValue();
616+
if (logger.isTraceEnabled())
617+
logger.trace("Storing the key to state {}({}) with value of {}", state.getKey(), state.getKey().ordinal(), vv);
618+
switch (state.getKey().ordinal())
619+
{
620+
case 15:
621+
filteredStates.put(ApplicationState.NATIVE_ADDRESS_AND_PORT, vv);
622+
break;
623+
case 16:
624+
break;
625+
case 17:
626+
filteredStates.put(ApplicationState.INTERNAL_ADDRESS_AND_PORT, vv);
627+
break;
628+
case 18:
629+
case 19:
630+
case 20:
631+
break;
632+
case 21:
633+
filteredStates.put(ApplicationState.DISK_USAGE, vv);
634+
break;
635+
default:
636+
filteredStates.put(state.getKey(), vv);
637+
}
638+
}
639+
return filteredStates;
640+
}
641+
return states;
642+
}
542643
}

‎src/java/org/apache/cassandra/net/OutboundConnectionInitiator.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -345,6 +345,12 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
345345
int peerMessagingVersion = msg.maxMessagingVersion;
346346
logger.trace("received second handshake message from peer {}, msg = {}", settings.connectTo, msg);
347347

348+
logger.debug("Summary of messaging versions while connecting to {} " +
349+
"peer messaging version {} request messaging version {} " +
350+
"max accept version {} min accept version {}",
351+
settings.connectTo, peerMessagingVersion, settings.acceptVersions.max,
352+
settings.acceptVersions.max, settings.acceptVersions.min);
353+
348354
FrameEncoder frameEncoder = null;
349355
Result<SuccessType> result;
350356
assert useMessagingVersion > 0;
@@ -379,6 +385,8 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
379385
}
380386
}
381387

388+
logger.debug("Result of connection initiation for {} is {}", settings.connectTo, result.outcome);
389+
382390
ChannelPipeline pipeline = ctx.pipeline();
383391
if (result.isSuccess())
384392
{

‎src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,14 @@ private void sendPingMessages(Set<InetAddressAndPort> peers, Map<String, CountDo
209209
Message<PingRequest> large = Message.out(PING_REQ, PingRequest.forLarge);
210210
for (InetAddressAndPort peer : peers)
211211
{
212+
boolean known = MessagingService.instance().versions.knows(peer);
213+
logger.debug("Peer {} is known with version {}", peer, known ? MessagingService.instance().versions.getRaw(peer) : "null");
214+
// DSE 6.8/6.9 advertises itself with value higher than VERSION_40, thus we need to compare it with VERSION_DSE_68
215+
boolean detectedDse = known && MessagingService.instance().versions.getRaw(peer) >= MessagingService.VERSION_DSE_68;
216+
if (MessagingService.instance().versions.get(peer) < MessagingService.VERSION_40 || detectedDse)
217+
// DSE 6.x doesn't support PING_REQ, and while C* 3.x does, PING only improves rolling restart times. CASSANDRA-13993 → CASSANDRA-14447
218+
continue;
219+
212220
MessagingService.instance().sendWithCallback(small, peer, responseHandler, SMALL_MESSAGES);
213221
MessagingService.instance().sendWithCallback(large, peer, responseHandler, LARGE_MESSAGES);
214222
}

‎src/java/org/apache/cassandra/schema/SchemaTransformations.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,9 @@
2020

2121
import java.util.Optional;
2222

23+
import org.slf4j.Logger;
24+
import org.slf4j.LoggerFactory;
25+
2326
import org.apache.cassandra.db.marshal.UserType;
2427
import org.apache.cassandra.exceptions.AlreadyExistsException;
2528
import org.apache.cassandra.exceptions.ConfigurationException;
@@ -31,6 +34,8 @@
3134
*/
3235
public class SchemaTransformations
3336
{
37+
private static final Logger logger = LoggerFactory.getLogger(SchemaTransformations.class);
38+
3439
/**
3540
* Creates a schema transformation that adds the provided keyspace.
3641
*
@@ -227,6 +232,22 @@ public Keyspaces apply(Keyspaces schema)
227232
updatedKeyspace = updatedKeyspace.withSwapped(updatedKeyspace.tables.with(updatedBuilder.build()));
228233
}
229234
}
235+
236+
if (curKeyspace.types != null)
237+
{
238+
for (UserType currType : curKeyspace.types)
239+
{
240+
UserType desiredType = updatedKeyspace.types.getNullable(currType.name);
241+
// if the type exist we keep the existing definition, otherwise we inherit the type
242+
// the motivation behind it is that there might be tables (inherited above) that depend on it
243+
if (desiredType == null)
244+
{
245+
logger.debug("Preserving type {} for keyspace {}", currType.getNameAsString(), curKeyspace.name);
246+
updatedKeyspace = updatedKeyspace.withSwapped(updatedKeyspace.types.with(currType));
247+
}
248+
}
249+
}
250+
230251
}
231252
return schema.withAddedOrUpdated(updatedKeyspace);
232253
}

0 commit comments

Comments
 (0)