Merge pull request #24 from terascope/fix-2.1.2

v2.1.3 Fix case when using no key and elasticsearch bulk
terascope · May 9, 2019 · a2c236e · a2c236e
2 parents eff234a + 816f056
commit a2c236e
Show file tree

Hide file tree

Showing 9 changed files with 12 additions and 11 deletions.
diff --git a/README.md b/README.md
@@ -160,7 +160,7 @@ terafoundation:
 {
     // make sure to include the asset bundle
     // additionally you can specify the version
-    // "kafka:2.1.2"
+    // "kafka:2.1.3"
     "assets": [ "kafka" ],
     "apis": [
         {
@@ -219,7 +219,7 @@ terafoundation:
     "slicers": 1,
     // make sure to include the asset bundle
     // additionally you can specify the version
-    // "kafka:2.1.2"
+    // "kafka:2.1.3"
     "assets": [ "kafka" ],
     // ...
     "operations": [
@@ -271,7 +271,7 @@ terafoundation:
 {
     // make sure to include the asset bundle
     // additionally you can specify the version
-    // "kafka:2.1.2"
+    // "kafka:2.1.3"
     "assets": [ "kafka" ],
     // ...
     "operations": [

diff --git a/asset/asset.json b/asset/asset.json
@@ -1,5 +1,5 @@
 {
     "name": "kafka",
     "description": "Kafka reader and writer support.",
-    "version": "2.1.2"
+    "version": "2.1.3"
 }
diff --git a/asset/package.json b/asset/package.json
@@ -1,6 +1,6 @@
 {
     "name": "kafka-assets",
-    "version": "2.1.2",
+    "version": "2.1.3",
     "private": true,
     "description": "Teraslice asset for kafka operations",
     "main": "index.js",

diff --git a/asset/src/_kafka_helpers/index.ts b/asset/src/_kafka_helpers/index.ts
@@ -54,7 +54,7 @@ function getErrorCause(err: any): string {
 
 export interface KafkaMessageMetadata {
     /** the message key */
-    _key: string;
+    _key: string|null;
     /** The time at which the data was ingested into the source data */
     _ingestTime: number;
     /** The time at which the data was consumed by the reader */

diff --git a/asset/src/kafka_reader/fetcher.ts b/asset/src/kafka_reader/fetcher.ts
@@ -41,6 +41,7 @@ export default class KafkaFetcher extends Fetcher<KafkaReaderConfig> {
     async fetch() {
         const map = this.tryRecord((msg: KafkaMessage): DataEntity => {
             const now = Date.now();
+
             const metadata: KafkaMessageMetadata = {
                 _key: keyToString(msg.key),
                 _ingestTime: msg.timestamp,
@@ -117,6 +118,6 @@ export default class KafkaFetcher extends Fetcher<KafkaReaderConfig> {
 
 /** Safely convert a buffer or string to a string */
 function keyToString(str?: string|Buffer) {
-    if (!str) return '';
+    if (!str) return null;
     return str.toString('utf8');
 }
diff --git a/docs/kafka_dead_letter/USAGE.md b/docs/kafka_dead_letter/USAGE.md
@@ -4,7 +4,7 @@
 {
     // make sure to include the asset bundle
     // additionally you can specify the version
-    // "kafka:2.1.2"
+    // "kafka:2.1.3"
     "assets": [ "kafka" ],
     "apis": [
         {

diff --git a/docs/kafka_reader/USAGE.md b/docs/kafka_reader/USAGE.md
@@ -6,7 +6,7 @@
     "slicers": 1,
     // make sure to include the asset bundle
     // additionally you can specify the version
-    // "kafka:2.1.2"
+    // "kafka:2.1.3"
     "assets": [ "kafka" ],
     // ...
     "operations": [

diff --git a/docs/kafka_sender/USAGE.md b/docs/kafka_sender/USAGE.md
@@ -4,7 +4,7 @@
 {
     // make sure to include the asset bundle
     // additionally you can specify the version
-    // "kafka:2.1.2"
+    // "kafka:2.1.3"
     "assets": [ "kafka" ],
     // ...
     "operations": [

diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
     "name": "kafka-asset-bundle",
-    "version": "2.1.2",
+    "version": "2.1.3",
     "description": "A bundle of Kafka operations and processors for Teraslice",
     "private": true,
     "scripts": {