From 66d7e4ebc123676ae0e8a42140e2581a93f3fc2f Mon Sep 17 00:00:00 2001
From: Sebastian Imlay <sebastian.imlay@gmail.com>
Date: Fri, 5 Feb 2021 13:42:48 -0800
Subject: [PATCH] Use encode_varint for vector size

---
 Cargo.lock                            | 26 +++++++++++------------
 VERSION                               |  2 +-
 src/auth/Cargo.toml                   |  6 +++---
 src/cli/Cargo.toml                    |  8 +++----
 src/client/Cargo.toml                 |  8 +++----
 src/client/src/lib.rs                 |  2 +-
 src/cluster/Cargo.toml                |  8 +++----
 src/controlplane-metadata/Cargo.toml  |  4 ++--
 src/controlplane/Cargo.toml           |  6 +++---
 src/dataplane-protocol/Cargo.toml     |  2 +-
 src/dataplane-protocol/src/batch.rs   |  2 +-
 src/dataplane-protocol/src/record.rs  | 30 +++++++++++++++++++--------
 src/extension-common/Cargo.toml       |  4 ++--
 src/extension-consumer/Cargo.toml     |  8 +++----
 src/extension-runner-local/Cargo.toml |  6 +++---
 src/sc-schema/Cargo.toml              |  6 +++---
 src/sc/Cargo.toml                     | 12 +++++------
 src/spu-schema/Cargo.toml             |  4 ++--
 src/spu/Cargo.toml                    | 12 +++++------
 src/storage/Cargo.toml                |  4 ++--
 src/storage/src/segment.rs            | 18 ++++++++--------
 tests/runner/Cargo.toml               |  2 +-
 22 files changed, 96 insertions(+), 84 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 133fe141672..c836baee21b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1117,7 +1117,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio"
-version = "0.4.1"
+version = "0.5.0"
 dependencies = [
  "async-channel",
  "async-mutex",
@@ -1148,7 +1148,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-auth"
-version = "0.3.0"
+version = "0.4.0"
 dependencies = [
  "async-trait",
  "fluvio-controlplane-metadata",
@@ -1170,7 +1170,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-cli"
-version = "0.4.1"
+version = "0.5.0"
 dependencies = [
  "async-h1",
  "color-eyre",
@@ -1265,7 +1265,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-controlplane"
-version = "0.5.0"
+version = "0.6.0"
 dependencies = [
  "fluvio-controlplane-metadata",
  "fluvio-dataplane-protocol",
@@ -1277,7 +1277,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-controlplane-metadata"
-version = "0.5.0"
+version = "0.6.0"
 dependencies = [
  "async-trait",
  "fluvio-dataplane-protocol",
@@ -1293,7 +1293,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-dataplane-protocol"
-version = "0.2.0"
+version = "0.3.0"
 dependencies = [
  "bytes 0.5.6",
  "content_inspector",
@@ -1310,7 +1310,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-extension-common"
-version = "0.1.0"
+version = "0.2.0"
 dependencies = [
  "async-trait",
  "fluvio",
@@ -1462,7 +1462,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-runner-local"
-version = "0.2.0"
+version = "0.3.0"
 dependencies = [
  "fluvio-future",
  "fluvio-sc",
@@ -1473,7 +1473,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-sc"
-version = "0.4.3"
+version = "0.5.0"
 dependencies = [
  "async-channel",
  "async-lock 1.1.5",
@@ -1514,7 +1514,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-sc-schema"
-version = "0.5.0"
+version = "0.6.0"
 dependencies = [
  "fluvio-controlplane-metadata",
  "fluvio-dataplane-protocol",
@@ -1571,7 +1571,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-spu"
-version = "0.3.3"
+version = "0.4.0"
 dependencies = [
  "async-channel",
  "async-io",
@@ -1609,7 +1609,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-spu-schema"
-version = "0.3.0"
+version = "0.4.0"
 dependencies = [
  "bytes 0.5.6",
  "fluvio-dataplane-protocol",
@@ -1622,7 +1622,7 @@ dependencies = [
 
 [[package]]
 name = "fluvio-storage"
-version = "0.2.1"
+version = "0.3.0"
 dependencies = [
  "async-channel",
  "async-mutex",
diff --git a/VERSION b/VERSION
index 0175995fc9f..ebcc7749ec7 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-0.7.0-alpha.4
\ No newline at end of file
+0.7.0-alpha.5
\ No newline at end of file
diff --git a/src/auth/Cargo.toml b/src/auth/Cargo.toml
index bd69edba236..e7c9ede40b5 100644
--- a/src/auth/Cargo.toml
+++ b/src/auth/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-auth"
-version = "0.3.0"
+version = "0.4.0"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 edition = "2018"
 license = "Apache-2.0"
@@ -14,8 +14,8 @@ path = "src/lib.rs"
 
 [dependencies]
 async-trait = "0.1.41"
-fluvio-controlplane-metadata = { version = "0.5.0", path = "../controlplane-metadata" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+fluvio-controlplane-metadata = { version = "0.6.0", path = "../controlplane-metadata" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
 fluvio-future = { version = "0.1.12", features = ["net", "openssl_tls"] }
 fluvio-protocol = { path = "../protocol",  version = "0.3.0" }
 fluvio-socket = { path = "../socket", version = "0.5.0" }
diff --git a/src/cli/Cargo.toml b/src/cli/Cargo.toml
index 89663186916..0dbfc054994 100644
--- a/src/cli/Cargo.toml
+++ b/src/cli/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-cli"
-version = "0.4.1"
+version = "0.5.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio CLI"
@@ -48,12 +48,12 @@ sysinfo = "0.16.1"
 k8-config = { version = "1.3.0" }
 k8-client = { version = "5.0.0" }
 fluvio-future = { version = "0.1.8", features = ["fs", "io", "subscriber"] }
-fluvio = { version = "0.4.1", path = "../client", default-features = false }
+fluvio = { version = "0.5.0", path = "../client", default-features = false }
 fluvio-cluster = { version = "0.7.0", path = "../cluster", default-features = false, features = ["cli"] }
 fluvio-package-index = { version = "0.2.0", path = "../package-index" }
 fluvio-extension-consumer = { version = "0.2.0", path = "../extension-consumer" }
-fluvio-extension-common = { version = "0.1.0", path = "../extension-common", features = ["target"]}
-fluvio-controlplane-metadata = { version = "0.5.0", path = "../controlplane-metadata", features = ["use_serde", "k8"] }
+fluvio-extension-common = { version = "0.2.0", path = "../extension-common", features = ["target"]}
+fluvio-controlplane-metadata = { version = "0.6.0", path = "../controlplane-metadata", features = ["use_serde", "k8"] }
 k8-types = { version = "0.1.0", features = ["core"]}
 
 [dev-dependencies]
diff --git a/src/client/Cargo.toml b/src/client/Cargo.toml
index 4f85603fc71..225175bcec4 100644
--- a/src/client/Cargo.toml
+++ b/src/client/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio"
-version = "0.4.1"
+version = "0.5.0"
 edition = "2018"
 license = "Apache-2.0"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
@@ -38,11 +38,11 @@ semver = "0.11.0"
 # Fluvio dependencies
 fluvio-future = { version = "0.1.15", features = ["task", "native2_tls"] }
 fluvio-types = { version = "0.2.0", path = "../types" }
-fluvio-sc-schema = { version = "0.5.0", path = "../sc-schema", default-features = false }
-fluvio-spu-schema = { version = "0.3.0", path = "../spu-schema" }
+fluvio-sc-schema = { version = "0.6.0", path = "../sc-schema", default-features = false }
+fluvio-spu-schema = { version = "0.4.0", path = "../spu-schema" }
 fluvio-socket = { path = "../socket", version = "0.5.0", features = ["tls"] }
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
 
 [dev-dependencies]
 async-std = "1.6.4"
diff --git a/src/client/src/lib.rs b/src/client/src/lib.rs
index 28a2556e468..f778adb2ea8 100644
--- a/src/client/src/lib.rs
+++ b/src/client/src/lib.rs
@@ -102,7 +102,7 @@ pub use crate::admin::FluvioAdmin;
 pub use crate::client::Fluvio;
 
 /// The minimum VERSION of the Fluvio Platform that this client is compatible with.
-const MINIMUM_PLATFORM_VERSION: &str = "0.7.0-alpha.1";
+const MINIMUM_PLATFORM_VERSION: &str = "0.7.0-alpha.5";
 
 /// Creates a producer that sends events to the named topic
 ///
diff --git a/src/cluster/Cargo.toml b/src/cluster/Cargo.toml
index 1cc6775e224..0aaaf4a2890 100644
--- a/src/cluster/Cargo.toml
+++ b/src/cluster/Cargo.toml
@@ -42,13 +42,13 @@ proclist = "0.9.2"
 remoteprocess = "0.4.2"
 
 # Fluvio dependencies
-fluvio = { version = "0.4.1", path = "../client", default-features = false }
+fluvio = { version = "0.5.0", path = "../client", default-features = false }
 fluvio-helm = "0.4.1"
 fluvio-future = { version = "0.1.13" }
 fluvio-command = { version = "0.2.0", path = "../command" }
-fluvio-runner-local = { version = "0.2.0", path = "../extension-runner-local", optional = true }
-fluvio-extension-common = { version = "0.1.0", path = "../extension-common", optional = true }
-fluvio-controlplane-metadata = { version = "0.5.0", path = "../controlplane-metadata", features = ["k8"] }
+fluvio-runner-local = { version = "0.3.0", path = "../extension-runner-local", optional = true }
+fluvio-extension-common = { version = "0.2.0", path = "../extension-common", optional = true }
+fluvio-controlplane-metadata = { version = "0.6.0", path = "../controlplane-metadata", features = ["k8"] }
 flv-util = "0.5.2"
 k8-config = { version = "1.3.0" }
 k8-client = { version = "5.0.0" }
diff --git a/src/controlplane-metadata/Cargo.toml b/src/controlplane-metadata/Cargo.toml
index 9145ee8d3e8..c3a0f187976 100644
--- a/src/controlplane-metadata/Cargo.toml
+++ b/src/controlplane-metadata/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "fluvio-controlplane-metadata"
 edition = "2018"
-version = "0.5.0"
+version = "0.6.0"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio metadata"
 repository = "https://github.com/infinyon/fluvio"
@@ -27,7 +27,7 @@ flv-util = { version = "0.5.0" }
 fluvio-types = { version = "0.2.0", path = "../types" }
 fluvio-stream-model = { path = "../stream-model", version = "0.4.0" }
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
 
 
 [dev-dependencies]
diff --git a/src/controlplane/Cargo.toml b/src/controlplane/Cargo.toml
index b3447dc057b..3d347f04eff 100644
--- a/src/controlplane/Cargo.toml
+++ b/src/controlplane/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-controlplane"
-version = "0.5.0"
+version = "0.6.0"
 edition = "2018"
 license = "Apache-2.0"
 description = "Fluvio control plane API"
@@ -17,6 +17,6 @@ tracing = "0.1.19"
 
 # Fluvio dependencies
 fluvio-types = { path = "../types", version = "0.2.0" }
-fluvio-controlplane-metadata = { path = "../controlplane-metadata", version = "0.5.0" }
+fluvio-controlplane-metadata = { path = "../controlplane-metadata", version = "0.6.0" }
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
diff --git a/src/dataplane-protocol/Cargo.toml b/src/dataplane-protocol/Cargo.toml
index a779950b458..e564485132b 100644
--- a/src/dataplane-protocol/Cargo.toml
+++ b/src/dataplane-protocol/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-dataplane-protocol"
-version = "0.2.0"
+version = "0.3.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "data plane protocol"
diff --git a/src/dataplane-protocol/src/batch.rs b/src/dataplane-protocol/src/batch.rs
index d51e3ee3ec1..04ffdf7cbe8 100644
--- a/src/dataplane-protocol/src/batch.rs
+++ b/src/dataplane-protocol/src/batch.rs
@@ -243,7 +243,7 @@ mod test {
 
         let decoded_record = batch.records.get(0).unwrap();
         println!("record crc: {}", batch.header.crc);
-        assert_eq!(batch.header.crc, 2908671645);
+        assert_eq!(batch.header.crc, 843514105);
         let b = decoded_record.value.as_ref();
         assert_eq!(b, b"test");
 
diff --git a/src/dataplane-protocol/src/record.rs b/src/dataplane-protocol/src/record.rs
index 2c82ba37ea4..c6c50e23771 100644
--- a/src/dataplane-protocol/src/record.rs
+++ b/src/dataplane-protocol/src/record.rs
@@ -10,8 +10,8 @@ use log::{trace, warn};
 use once_cell::sync::Lazy;
 
 use crate::core::bytes::Buf;
-use crate::core::bytes::BufExt;
 use crate::core::bytes::BufMut;
+use bytes::buf::BufExt;
 
 use crate::core::Decoder;
 use crate::core::DecoderVarInt;
@@ -138,14 +138,22 @@ impl<'a> From<&'a [u8]> for DefaultAsyncBuffer {
 
 impl Encoder for DefaultAsyncBuffer {
     fn write_size(&self, version: Version) -> usize {
-        self.0.write_size(version)
+        let len = self.0.len() as i64;
+        self.0.iter().fold(len.var_write_size(), |sum, val| {
+            sum + val.write_size(version)
+        })
     }
 
-    fn encode<T>(&self, src: &mut T, version: Version) -> Result<(), Error>
+    fn encode<T>(&self, dest: &mut T, version: Version) -> Result<(), Error>
     where
         T: BufMut,
     {
-        self.0.encode(src, version)
+        let len: i64 = self.0.len() as i64;
+        len.encode_varint(dest)?;
+        for v in self.0.iter() {
+            v.encode(dest, version)?;
+        }
+        Ok(())
     }
 }
 
@@ -156,7 +164,14 @@ impl Decoder for DefaultAsyncBuffer {
     {
         trace!("decoding default asyncbuffer");
         if let Some(ref mut buffer) = Arc::get_mut(&mut self.0) {
-            buffer.decode(src, version)
+            let mut len: i64 = 0;
+            len.decode_varint(src)?;
+            for _ in 0..len {
+                let mut value = <u8>::default();
+                value.decode(src, version)?;
+                buffer.push(value);
+            }
+            Ok(())
         } else {
             Err(Error::new(
                 ErrorKind::Other,
@@ -434,10 +449,7 @@ mod test {
         record.encode(&mut out, 0)?;
         println!("ENCODED: {:#x?}", out);
         */
-
-        let data = [
-            0x1e, 0x0, 0x0, 0x2, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x3, 0x64, 0x6f, 0x67, 0x0,
-        ];
+        let data = [0x12, 0x0, 0x0, 0x2, 0x0, 0x6, 0x64, 0x6f, 0x67, 0x0];
 
         let record = DefaultRecord::decode_from(&mut Cursor::new(&data), 0)?;
         assert_eq!(record.as_bytes(0)?.len(), data.len());
diff --git a/src/extension-common/Cargo.toml b/src/extension-common/Cargo.toml
index 04ec3203e34..69c7bf44c09 100644
--- a/src/extension-common/Cargo.toml
+++ b/src/extension-common/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-extension-common"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio extension common"
@@ -28,4 +28,4 @@ futures-lite = { version = "1.7.0" }
 thiserror = "1.0.20"
 
 
-fluvio = { version = "0.4.0", path = "../client",  optional = true }
+fluvio = { version = "0.5.0", path = "../client",  optional = true }
diff --git a/src/extension-consumer/Cargo.toml b/src/extension-consumer/Cargo.toml
index fed07a2fa73..dbc27b5e8b5 100644
--- a/src/extension-consumer/Cargo.toml
+++ b/src/extension-consumer/Cargo.toml
@@ -30,8 +30,8 @@ hostname-validator = "1.0.0"
 
 flv-util = { version = "0.5.0" }
 fluvio-future = { version = "0.1.8", features = ["fs", "io"] }
-fluvio = { version = "0.4.1", path = "../client", default-features = false }
+fluvio = { version = "0.5.0", path = "../client", default-features = false }
 fluvio-types = { version = "0.2.0" , path = "../types" }
-fluvio-extension-common = { version = "0.1.0", path = "../extension-common", features = ["target"] }
-fluvio-controlplane-metadata = { version = "0.5.0", path = "../controlplane-metadata", features = ["use_serde"] }
-fluvio-sc-schema = { version = "0.5.0", path = "../sc-schema", features = ["use_serde"] }
+fluvio-extension-common = { version = "0.2.0", path = "../extension-common", features = ["target"] }
+fluvio-controlplane-metadata = { version = "0.6.0", path = "../controlplane-metadata", features = ["use_serde"] }
+fluvio-sc-schema = { version = "0.6.0", path = "../sc-schema", features = ["use_serde"] }
diff --git a/src/extension-runner-local/Cargo.toml b/src/extension-runner-local/Cargo.toml
index 6c3154c8c82..b46f73234ef 100644
--- a/src/extension-runner-local/Cargo.toml
+++ b/src/extension-runner-local/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-runner-local"
-version = "0.2.0"
+version = "0.3.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio Engine Runner"
@@ -22,5 +22,5 @@ thiserror = "1.0.20"
 
 # regardless of TLS, sc and spu always use openssl_tls for now because we need cert API
 fluvio-future = { version = "0.1.12", features = ["subscriber"] }
-fluvio-sc = { version = "0.4.0", path = "../sc" }
-fluvio-spu = { version = "0.3.0", path = "../spu" }
+fluvio-sc = { version = "0.5.0", path = "../sc" }
+fluvio-spu = { version = "0.4.0", path = "../spu" }
diff --git a/src/sc-schema/Cargo.toml b/src/sc-schema/Cargo.toml
index fc5114518f6..bdc9b215d85 100644
--- a/src/sc-schema/Cargo.toml
+++ b/src/sc-schema/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-sc-schema"
-version = "0.5.0"
+version = "0.6.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio API for SC"
@@ -23,6 +23,6 @@ static_assertions = "1.1.0"
 
 # Fluvio dependencies
 fluvio-types = { version = "0.2.0", path = "../types" }
-fluvio-controlplane-metadata = { version = "0.5.0", default-features = false, path = "../controlplane-metadata" }
+fluvio-controlplane-metadata = { version = "0.6.0", default-features = false, path = "../controlplane-metadata" }
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
diff --git a/src/sc/Cargo.toml b/src/sc/Cargo.toml
index 710f43c0435..90f031a6314 100644
--- a/src/sc/Cargo.toml
+++ b/src/sc/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "fluvio-sc"
 edition = "2018"
-version = "0.4.3"
+version = "0.5.0"
 authors = ["fluvio.io"]
 description = "Fluvio Stream Controller"
 repository = "https://github.com/infinyon/fluvio"
@@ -40,19 +40,19 @@ semver = "0.11.0"
 once_cell = "1.5"
 
 # Fluvio dependencies
-fluvio-auth = { version = "0.3.0", path = "../auth" }
+fluvio-auth = { version = "0.4.0", path = "../auth" }
 fluvio-future = { version = "0.1.12", features = ["subscriber", "openssl_tls"] }
 fluvio-types = { version = "0.2.0", path = "../types", features = ["events"] }
-fluvio-sc-schema = { version = "0.5.0", path = "../sc-schema" }
+fluvio-sc-schema = { version = "0.6.0", path = "../sc-schema" }
 fluvio-stream-model = { path = "../stream-model", version = "0.4.0" }
-fluvio-controlplane = { version = "0.5.0", path = "../controlplane" }
-fluvio-controlplane-metadata = { version = "0.5.0", features = ["k8", "serde"], path = "../controlplane-metadata" }
+fluvio-controlplane = { version = "0.6.0", path = "../controlplane" }
+fluvio-controlplane-metadata = { version = "0.6.0", features = ["k8", "serde"], path = "../controlplane-metadata" }
 fluvio-stream-dispatcher = { version = "0.4.0", path = "../stream-dispatcher" }
 k8-client = { version = "5.0.0", optional = true }
 k8-metadata-client = { version = "3.0.0" }
 k8-types = { version = "0.1.0", features = ["app"]}
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
 fluvio-socket = { path = "../socket", version = "0.5.0" }
 fluvio-service = { path = "../service", version = "0.4.0" }
 flv-tls-proxy = { version = "0.3.0" }
diff --git a/src/spu-schema/Cargo.toml b/src/spu-schema/Cargo.toml
index 262ec5d9c1e..5583222709d 100644
--- a/src/spu-schema/Cargo.toml
+++ b/src/spu-schema/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-spu-schema"
-version = "0.3.0"
+version = "0.4.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio API for SPU"
@@ -20,4 +20,4 @@ static_assertions = "1.1.0"
 
 # Fluvio dependencies
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
diff --git a/src/spu/Cargo.toml b/src/spu/Cargo.toml
index 533f8f3cbca..689af0ce0f4 100644
--- a/src/spu/Cargo.toml
+++ b/src/spu/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "fluvio-spu"
 edition = "2018"
-version = "0.3.3"
+version = "0.4.0"
 authors = ["fluvio.io"]
 description = "Fluvio Stream Processing Unit"
 repository = "https://github.com/infinyon/fluvio"
@@ -40,12 +40,12 @@ async-io = "1.3.1"
 
 # Fluvio dependencies
 fluvio-types = { version = "0.2.0", path = "../types" }
-fluvio-storage = { version = "0.2.0", path = "../storage" }
-fluvio-controlplane = { version = "0.5.0", path = "../controlplane" }
-fluvio-controlplane-metadata = { version = "0.5.0", path = "../controlplane-metadata" }
-fluvio-spu-schema = { version = "0.3.0", path = "../spu-schema" }
+fluvio-storage = { version = "0.3.0", path = "../storage" }
+fluvio-controlplane = { version = "0.6.0", path = "../controlplane" }
+fluvio-controlplane-metadata = { version = "0.6.0", path = "../controlplane-metadata" }
+fluvio-spu-schema = { version = "0.4.0", path = "../spu-schema" }
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
 fluvio-socket = { path = "../socket", version = "0.5.0" }
 fluvio-service = { path = "../service", version = "0.4.0" }
 flv-tls-proxy = { version = "0.3.0" }
diff --git a/src/storage/Cargo.toml b/src/storage/Cargo.toml
index 230781f625f..d2de8169263 100644
--- a/src/storage/Cargo.toml
+++ b/src/storage/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 edition = "2018"
 name = "fluvio-storage"
-version = "0.2.1"
+version = "0.3.0"
 authors = ["fluvio.io"]
 description = "Storage for Fluvio platform"
 repository = "https://github.com/infinyon/fluvio"
@@ -31,7 +31,7 @@ serde = { version = "1.0.103", features = ['derive'] }
 fluvio-types = { version = "0.2.0" , path = "../types" }
 fluvio-future = { version = "0.1.8", features = ["fs","mmap"] }
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
 async-mutex = "1.4.0"
 
 
diff --git a/src/storage/src/segment.rs b/src/storage/src/segment.rs
index e046303a053..9c8bfa502be 100644
--- a/src/storage/src/segment.rs
+++ b/src/storage/src/segment.rs
@@ -453,7 +453,7 @@ mod tests {
             (active_segment.find_offset_position(20).await?).expect("offset exists");
         assert_eq!(offset_position.get_base_offset(), 20);
         assert_eq!(offset_position.get_pos(), 0); //
-        assert_eq!(offset_position.len(), 64);
+        assert_eq!(offset_position.len(), 58);
         assert!((active_segment.find_offset_position(30).await?).is_none());
         Ok(())
     }
@@ -492,7 +492,7 @@ mod tests {
             (active_segment.find_offset_position(20).await?).expect("offset exists");
         assert_eq!(offset_position.get_base_offset(), 20);
         assert_eq!(offset_position.get_pos(), 0); //
-        assert_eq!(offset_position.len(), 109);
+        assert_eq!(offset_position.len(), 85);
         assert!((active_segment.find_offset_position(30).await?).is_none());
 
         Ok(())
@@ -516,10 +516,10 @@ mod tests {
 
         assert_eq!(seg_sink.get_end_offset(), 46);
 
-        assert_eq!(seg_sink.get_log_pos(), 273); // each takes 91 bytes
+        assert_eq!(seg_sink.get_log_pos(), 237); // each takes 79 bytes
 
         let index = seg_sink.get_index();
-        assert_eq!(index[0].to_be(), (2, 91));
+        assert_eq!(index[0].to_be(), (2, 79));
 
         let bytes = read_bytes_from_file(&test_dir.join(TEST2_FILE_NAME))?;
         debug!("read {} bytes", bytes.len());
@@ -536,16 +536,16 @@ mod tests {
         let offset_pos1 = seg_sink.find_offset_position(40).await?.expect("pos");
         assert_eq!(offset_pos1.get_base_offset(), 40);
         assert_eq!(offset_pos1.get_pos(), 0);
-        assert_eq!(offset_pos1.len(), 79);
+        assert_eq!(offset_pos1.len(), 67);
         let offset_pos2 = seg_sink.find_offset_position(42).await?.expect("pos");
         assert_eq!(offset_pos2.get_base_offset(), 42);
-        assert_eq!(offset_pos2.get_pos(), 91);
-        assert_eq!(offset_pos2.len(), 79);
+        assert_eq!(offset_pos2.get_pos(), 79);
+        assert_eq!(offset_pos2.len(), 67);
 
         let offset_pos3 = seg_sink.find_offset_position(44).await?.expect("pos");
         assert_eq!(offset_pos3.get_base_offset(), 44);
-        assert_eq!(offset_pos3.get_pos(), 182);
-        assert_eq!(offset_pos3.len(), 79);
+        assert_eq!(offset_pos3.get_pos(), 158);
+        assert_eq!(offset_pos3.len(), 67);
 
         // test whether you can send batch with non zero base offset
         let mut next_batch = create_batch();
diff --git a/tests/runner/Cargo.toml b/tests/runner/Cargo.toml
index 5e82b951a50..a3afae3b702 100644
--- a/tests/runner/Cargo.toml
+++ b/tests/runner/Cargo.toml
@@ -18,5 +18,5 @@ fluvio-types = { version = "0.2.0", path = "../../src/types"  }
 fluvio-future = { version = "0.1.0", features = ["task","timer","subscriber","fixture"] }
 fluvio-command = { path = "../../src/command" }
 fluvio-controlplane-metadata = { features = ["k8"], path = "../../src/controlplane-metadata" }
-dataplane = { version = "0.2.0", path = "../../src/dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../../src/dataplane-protocol", package = "fluvio-dataplane-protocol" }
 fluvio-system-util = { version = "0.1.0", path = "../../src/utils" }