Merge pull request #23 from terascope/fixes-key

v2.1.2 return a key not a buffer in metadata, fixes #22

.travis.yml

@@ -27,7 +27,7 @@ before_install:
   - export KAFKA_TMP="$HOME/.kafka-downloads"
   - mkdir -p "$KAFKA_HOME"
   - mkdir -p "$KAFKA_TMP"
-  - '[[ ! -f "$KAFKA_TMP/kafka.tgz" ]] && wget https://www.apache.org/dist/kafka/2.1.0/kafka_2.11-2.1.0.tgz -O "$KAFKA_TMP/kafka.tgz" || echo "* using cache"'
+  - '[[ ! -f "$KAFKA_TMP/kafka.tgz" ]] && wget https://www.apache.org/dist/kafka/2.1.1/kafka_2.11-2.1.1.tgz -O "$KAFKA_TMP/kafka.tgz" || echo "* using cache"'
   - tar xzf "$KAFKA_TMP/kafka.tgz" -C $KAFKA_HOME --strip-components 1
   # add bin to path for ease of use
   - export PATH="$KAFKA_HOME/bin:$PATH"

README.md

@@ -160,7 +160,7 @@ terafoundation:
 {
     // make sure to include the asset bundle
     // additionally you can specify the version
-    // "kafka:2.1.1"
+    // "kafka:2.1.2"
     "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.1"
+    // "kafka:2.1.2"
     "assets": [ "kafka" ],
     // ...
     "operations": [
@@ -271,7 +271,7 @@ terafoundation:
 {
     // make sure to include the asset bundle
     // additionally you can specify the version
-    // "kafka:2.1.1"
+    // "kafka:2.1.2"
     "assets": [ "kafka" ],
     // ...
     "operations": [

asset/asset.json

@@ -1,5 +1,5 @@
 {
     "name": "kafka",
     "description": "Kafka reader and writer support.",
-    "version": "2.1.1"
+    "version": "2.1.2"
 }

asset/package.json

@@ -1,6 +1,6 @@
 {
     "name": "kafka-assets",
-    "version": "2.1.1",
+    "version": "2.1.2",
     "private": true,
     "description": "Teraslice asset for kafka operations",
     "main": "index.js",

asset/src/_kafka_clients/interfaces.ts

@@ -26,7 +26,7 @@ export interface ConsumerClientConfig {
 
 export interface ProduceMessage {
     data: Buffer;
-    key: string|null;
+    key: Buffer|string|null;
     timestamp: number|null;
 }
 

asset/src/_kafka_helpers/index.ts

@@ -4,7 +4,7 @@ import { codeToMessage, okErrors } from './error-codes';
 export type AnyKafkaError = Error|KafkaError|number|string|null;
 
 export interface KafkaError extends Error {
-    code: number;
+    code: number|string;
 }
 
 export function wrapError(message: string, err: AnyKafkaError): KafkaError {
@@ -34,7 +34,7 @@ function getErrorCause(err: any): string {
     let message = causedBy;
     message += typeof err === 'object' ? err.message : toString(err);
 
-    let code: number|null = null;
+    let code: number|string|null = null;
 
     if (isKafkaError(err)) {
         code = err.code;
@@ -73,7 +73,7 @@ export interface KafkaMessageMetadata {
 
 export interface KafkaMessage {
     /** the message key */
-    key: string;
+    key: Buffer|string;
     /** the topic name */
     topic: string;
     /** the partition on the topic the message was on */
@@ -93,7 +93,7 @@ export function isKafkaError(err: any): err is KafkaError {
     return isError(err) && err.code != null;
 }
 
-export function isOkayError(err: AnyKafkaError, action: string = 'any'): boolean {
+export function isOkayError(err?: AnyKafkaError, action: string = 'any'): boolean {
     if (err == null) return false;
     const code = isKafkaError(err) ? err.code : err as number;
     if (action === 'retryable') {

asset/src/kafka_reader/fetcher.ts

@@ -42,7 +42,7 @@ export default class KafkaFetcher extends Fetcher<KafkaReaderConfig> {
         const map = this.tryRecord((msg: KafkaMessage): DataEntity => {
             const now = Date.now();
             const metadata: KafkaMessageMetadata = {
-                _key: msg.key,
+                _key: keyToString(msg.key),
                 _ingestTime: msg.timestamp,
                 _processTime: now,
                 // TODO this should be based of an actual value
@@ -114,3 +114,9 @@ export default class KafkaFetcher extends Fetcher<KafkaReaderConfig> {
         return connection.client;
     }
 }
+
+/** Safely convert a buffer or string to a string */
+function keyToString(str?: string|Buffer) {
+    if (!str) return '';
+    return str.toString('utf8');
+}

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.1"
+    // "kafka:2.1.2"
     "assets": [ "kafka" ],
     "apis": [
         {

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.1"
+    // "kafka:2.1.2"
     "assets": [ "kafka" ],
     // ...
     "operations": [

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.1"
+    // "kafka:2.1.2"
     "assets": [ "kafka" ],
     // ...
     "operations": [

package.json

@@ -1,6 +1,6 @@
 {
     "name": "kafka-asset-bundle",
-    "version": "2.1.1",
+    "version": "2.1.2",
     "description": "A bundle of Kafka operations and processors for Teraslice",
     "private": true,
     "scripts": {

test/helpers-spec.ts

@@ -76,7 +76,7 @@ describe('isOkayError helper', () => {
     describe.each(alwaysBad)('when consuming and checking error code %s', (code) => {
         it('should return false', () => {
             const err = new Error('Uh oh') as KafkaError;
-            err.code = code;
+            err.code = code as string|number;
 
             expect(isOkayError(err, 'consume')).toBeFalse();
             expect(isOkayError(code, 'consume')).toBeFalse();