Repository: AlexPikalov/cdrs (language: rust)

Issue:
Panic when shutting cassandra down
Saw this today while restarting one of the two cassandra nodes I have CDRS cluster connected to.

```ERROR:cdrs::frame: Unexpected Cassandra version 0, either 4 or 132 is expected```

Not sure how it happened, but immediately after that the program panicked and quit.


Discussion / hints:
@bfrog 

Which Cassandra version do you use in your cluster? Did you use any compression? TLS connection?

Thanks
3.0.9, no compression

On Jun 24, 2017 3:14 AM, "Alex Pikalov" <notifications@github.com> wrote:

> @bfrog <https://github.com/bfrog>
>
> Which Cassandra version do you use in your cluster? Did you start session
> with any compression?
>
> Thanks
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/AlexPikalov/cdrs/issues/163#issuecomment-310824312>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/AAISn3wD4_S72XSzLsTdrSnVSe1QMi3Mks5sHMVcgaJpZM4OEEUO>
> .
>

TLS or plain TCP connection?
Plain

On Jun 24, 2017 3:39 PM, "Alex Pikalov" <notifications@github.com> wrote:

> TLS or plain TCP connection?
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/AlexPikalov/cdrs/issues/163#issuecomment-310864468>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/AAISn5wgI_6nmeJ-rPRYD2v3h-RII2Wlks5sHXPxgaJpZM4OEEUO>
> .
>

Thanks, will try with 3.0.9.
@bfrog  how many nodes were there to begin with and what was the replication factor ?
2 and 2, I restarted one and got that error.

On Sun, Jun 25, 2017 at 1:26 PM, harrydevnull <notifications@github.com>
wrote:

> @bfrog <https://github.com/bfrog> how many nodes were there to begin with
> and what was the replication factor ?
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/AlexPikalov/cdrs/issues/163#issuecomment-310919215>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/AAISn4vYYN_WUF9lzQRMZLGxqVTXK59Yks5sHqZIgaJpZM4OEEUO>
> .
>

Probably we need to revise a mechanism of node invalidation in [cluster](https://github.com/AlexPikalov/cdrs/blob/master/src/cluster.rs#L131). Let me please take a look closer on the case you've described.

Fix patch:
diff --git a/.travis.yml b/.travis.yml
index ee418b4..44a655b 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -13,8 +13,8 @@ before_install:
  - export JAVA_HOME=/usr/lib/jvm/java-8-oracle
  - java -version
  - sudo rm -rf /var/lib/cassandra/*
- - if [ ! -d "apache-cassandra-3.10" ]; then wget http://www.us.apache.org/dist/cassandra/3.10/apache-cassandra-3.10-bin.tar.gz && tar -xvzf apache-cassandra-3.10-bin.tar.gz; fi
- - sudo sh ./apache-cassandra-3.10/bin/cassandra -R
+ - if [ ! -d "apache-cassandra-3.9" ]; then wget https://archive.apache.org/dist/cassandra/3.9/apache-cassandra-3.9-bin.tar.gz && tar -xvzf apache-cassandra-3.9-bin.tar.gz; fi
+ - sudo sh ./apache-cassandra-3.9/bin/cassandra -R
  - sleep 20
 
 
diff --git a/src/client.rs b/src/client.rs
index fb06a26..0b9748f 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -52,12 +52,10 @@ impl<'a, T: Authenticator + 'a, X: CDRSTransport + 'a> CDRS<T, X> {
         try!(self.transport.write(options_frame.as_slice()));
 
         parse_frame(&mut self.transport, &self.compressor)
-        .map(|frame| {
-                match frame.get_body() {
-                Ok(ResponseBody::Supported(ref supported_body)) => supported_body.data.clone(),
-                _ => unreachable!(),
-            }
-        })
+            .map(|frame| match frame.get_body() {
+                     Ok(ResponseBody::Supported(ref supported_body)) => supported_body.data.clone(),
+                     _ => unreachable!(),
+                 })
     }
 
     /// The method establishes connection to the server which address was provided on previous
diff --git a/src/frame/parser.rs b/src/frame/parser.rs
index 7352e38..2f1789a 100644
--- a/src/frame/parser.rs
+++ b/src/frame/parser.rs
@@ -9,7 +9,6 @@ use types::data_serialization_types::decode_timeuuid;
 use error;
 
 pub fn parse_frame(mut cursor: &mut Read, compressor: &Compression) -> error::Result<Frame> {
-    // TODO: try to use slices instead
     let mut version_bytes = [0; VERSION_LEN];
     let mut flag_bytes = [0; FLAG_LEN];
     let mut opcode_bytes = [0; OPCODE_LEN];
@@ -17,11 +16,15 @@ pub fn parse_frame(mut cursor: &mut Read, compressor: &Compression) -> error::Re
     let mut length_bytes = [0; LENGTH_LEN];
 
     // NOTE: order of reads matters
-    try!(cursor.read(&mut version_bytes));
-    try!(cursor.read(&mut flag_bytes));
-    try!(cursor.read(&mut stream_bytes));
-    try!(cursor.read(&mut opcode_bytes));
-    try!(cursor.read(&mut length_bytes));
+    let v = try!(cursor.read(&mut version_bytes));
+    let f = try!(cursor.read(&mut flag_bytes));
+    let s = try!(cursor.read(&mut stream_bytes));
+    let o = try!(cursor.read(&mut opcode_bytes));
+    let l = try!(cursor.read(&mut length_bytes));
+
+    if v == 0 || f == 0 || s == 0 || o == 0 || l == 0 {
+        return Err(error::Error::from("Empty frame received"));
+    }
 
     let version = Version::from(version_bytes.to_vec());
     let flags = Flag::get_collection(flag_bytes[0]);
@@ -33,6 +36,7 @@ pub fn parse_frame(mut cursor: &mut Read, compressor: &Compression) -> error::Re
     unsafe {
         body_bytes.set_len(length);
     }
+
     try!(cursor.read_exact(&mut body_bytes));
 
     let full_body = if flags.iter().any(|flag| flag == &Flag::Compression) {
@@ -49,7 +53,6 @@ pub fn parse_frame(mut cursor: &mut Read, compressor: &Compression) -> error::Re
         unsafe {
             tracing_bytes.set_len(UUID_LEN);
         }
-        // TODO: try to use slice instead
         try!(body_cursor.read_exact(&mut tracing_bytes));
 
         decode_timeuuid(tracing_bytes.as_slice()).ok()
@@ -77,10 +80,10 @@ pub fn parse_frame(mut cursor: &mut Read, compressor: &Compression) -> error::Re
         warnings: warnings,
     };
 
-    conver_frame_into_result(frame)
+    convert_frame_into_result(frame)
 }
 
-fn conver_frame_into_result(frame: Frame) -> error::Result<Frame> {
+fn convert_frame_into_result(frame: Frame) -> error::Result<Frame> {
     match frame.opcode {
         Opcode::Error => {
             frame.get_body().and_then(|err| match err {


Test patch: