fixup: incorporate PR feedback on how to handle partition keys

Also, incorporates all of the changes from cloudevents#457

Signed-off-by: Lance Ball <[email protected]>

Author: lance

src/index.ts

@@ -8,7 +8,9 @@ import { ValidationError } from "./event/validation";
 import { CloudEventV1, CloudEventV1Attributes } from "./event/interfaces";
 
 import { Options, TransportFunction, EmitterFunction, emitterFor, Emitter } from "./transport/emitter";
-import { Headers, Mode, Binding, HTTP, Message, Serializer, Deserializer } from "./message";
+import { 
+  Headers, Mode, Binding, HTTP, Kafka, KafkaEvent, KafkaMessage, Message,
+  Serializer, Deserializer } from "./message";
 
 import CONSTANTS from "./constants";
 
@@ -27,6 +29,9 @@ export {
   Deserializer,
   Serializer,
   HTTP,
+  Kafka,
+  KafkaEvent,
+  KafkaMessage,
   // From transport
   TransportFunction,
   EmitterFunction,

src/message/kafka/headers.ts

@@ -3,7 +3,7 @@
  SPDX-License-Identifier: Apache-2.0
 */
 
-import { CloudEvent, CONSTANTS, Headers } from "../..";
+import { CloudEventV1, CONSTANTS, Headers } from "../..";
 
 type KafkaHeaders = Readonly<{
   ID: string;
@@ -55,7 +55,7 @@ export const HEADER_MAP: { [key: string]: string } = {
  * @param {CloudEvent} event a CloudEvent object
  * @returns {Headers} the CloudEvent attribute as Kafka headers
  */
-export function headersFor<T>(event: CloudEvent<T>): Headers {
+export function headersFor<T>(event: CloudEventV1<T>): Headers {
   const headers: Headers = {};
 
   Object.getOwnPropertyNames(event).forEach((property) => {

src/message/kafka/index.ts

@@ -8,6 +8,13 @@ import { Message, Headers, Binding } from "..";
 import { headersFor, HEADER_MAP, KAFKA_CE_HEADERS } from "./headers";
 import { sanitize } from "../http/headers";
 
+// Export the binding implementation and message interface
+export {
+  Kafka,
+  KafkaMessage,
+  KafkaEvent
+};
+
 /**
  * Bindings for Kafka transport
  * @implements {@linkcode Binding}
@@ -19,45 +26,39 @@ import { sanitize } from "../http/headers";
   isEvent: isKafkaEvent,
 };
 
+type Key = string | Buffer;
+
 /**
  * Extends the base Message type to include
  * Kafka-specific fields
  */
 interface KafkaMessage<T = string> extends Message {
-  key: string | Buffer
+  key: Key
   value: T | string | Buffer | unknown
   timestamp?: string
 }
 
 /**
- * Extends the base CloudEvent type to include
- * the Kafka-specific `partitionkey` field, which
- * is explicitly mapped to `KafkaMessage#key`.
+ * Extends the base CloudEventV1 interface to include a `partitionkey` field
+ * which is explicitly mapped to KafkaMessage#key
  */
 interface KafkaEvent<T> extends CloudEventV1<T> {
   /**
    * Maps to KafkaMessage#key per
    * https://github.com/cloudevents/spec/blob/v1.0.1/kafka-protocol-binding.md#31-key-mapping
    */
-  partitionkey: string | Buffer
+  partitionkey: Key
 }
 
-// Export the binding implementation and message interface
-export {
-  Kafka,
-  KafkaMessage,
-  KafkaEvent
-};
-
 /**
  * Serialize a CloudEvent for Kafka in binary mode
  * @implements {Serializer}
  * @see https://github.com/cloudevents/spec/blob/v1.0.1/kafka-protocol-binding.md#32-binary-content-mode
  *
- * @param {CloudEvent<T>} event The event to serialize
+ * @param {KafkaEvent<T>} event The event to serialize
  * @returns {KafkaMessage<T>} a KafkaMessage instance
  */
-function toBinaryKafkaMessage<T>(event: CloudEvent<T>): KafkaMessage<T> {
+function toBinaryKafkaMessage<T>(event: CloudEventV1<T>): KafkaMessage<T> {
   // 3.2.1. Content Type
   // For the binary mode, the header content-type property MUST be mapped directly
   // to the CloudEvents datacontenttype attribute.
@@ -67,7 +68,7 @@ function toBinaryKafkaMessage<T>(event: CloudEvent<T>): KafkaMessage<T> {
   };
   return {
     headers,
-    key: createKey(event),
+    key: event.partitionkey as Key,
     value: event.data,
     body: event.data,
     timestamp: timestamp(event.time)
@@ -82,14 +83,17 @@ function toBinaryKafkaMessage<T>(event: CloudEvent<T>): KafkaMessage<T> {
  * @param {CloudEvent<T>} event the CloudEvent to be serialized
  * @returns {KafkaMessage<T>} a KafkaMessage instance
  */
- function toStructuredKafkaMessage<T>(event: CloudEvent<T>): KafkaMessage<T> {
-  if (event.data_base64) {
+ function toStructuredKafkaMessage<T>(event: CloudEventV1<T>): KafkaMessage<T> {
+  if (event.data_base64 && event instanceof CloudEvent) {
     // The event's data is binary - delete it
     event = event.cloneWith({ data: undefined });
   }
   const value = event.toString();
   return {
-    key: createKey(event),
+    // All events may not have a partitionkey set, but if they do,
+    // use it for the KafkaMessage#key per
+    // https://github.com/cloudevents/spec/blob/v1.0.1/kafka-protocol-binding.md#31-key-mapping
+    key: event.partitionkey as Key,
     value,
     headers: {
       [CONSTANTS.HEADER_CONTENT_TYPE]: CONSTANTS.DEFAULT_CE_CONTENT_TYPE,
@@ -104,27 +108,28 @@ function toBinaryKafkaMessage<T>(event: CloudEvent<T>): KafkaMessage<T> {
  * @implements {Deserializer}
  *
  * @param {Message} message the incoming message
- * @return {CloudEvent} A new {CloudEvent} instance
+ * @return {KafkaEvent} A new {KafkaEvent} instance
  */
-function deserializeKafkaMessage<T>(message: Message): CloudEventV1<T> | CloudEvent<T> | CloudEvent<T>[] {
+function deserializeKafkaMessage<T>(message: Message): CloudEvent<T> | CloudEvent<T>[] {
   if (!isKafkaEvent(message)) {
     throw new ValidationError("No CloudEvent detected");
   }
-  if (!(message as KafkaMessage<T>).value) {
+  const m = message as KafkaMessage<T>;
+  if (!m.value) {
     throw new ValidationError("Value is null or undefined");
   }
-  if (!message.headers) {
+  if (!m.headers) {
     throw new ValidationError("Headers are null or undefined");
   }
-  const cleanHeaders: Headers = sanitize(message.headers);
+  const cleanHeaders: Headers = sanitize(m.headers);
   const mode: Mode = getMode(cleanHeaders);
   switch (mode) {
     case Mode.BINARY:
-      return parseBinary(message as KafkaMessage<T>) as CloudEventV1<T>;
+      return parseBinary(m);
     case Mode.STRUCTURED:
-      return parseStructured(message as KafkaMessage<T>);
+      return parseStructured(m);
     case Mode.BATCH:
-      return parseBatched(message as KafkaMessage<T>);
+      return parseBatched(m);
     default:
       throw new ValidationError("Unknown Message mode");
   }
@@ -137,22 +142,13 @@ function deserializeKafkaMessage<T>(message: Message): CloudEventV1<T> | CloudEv
  * @param {Message} message an incoming Message object
  * @returns {boolean} true if this Message is a CloudEvent
  */
-function isKafkaEvent<T>(message: Message<T>): boolean {
+function isKafkaEvent(message: Message): boolean {
    const headers = sanitize(message.headers);
    return !!headers[KAFKA_CE_HEADERS.ID] || // A binary mode event
     headers[CONSTANTS.HEADER_CONTENT_TYPE]?.startsWith(CONSTANTS.MIME_CE) as boolean || // A structured mode event
     headers[CONSTANTS.HEADER_CONTENT_TYPE]?.startsWith(CONSTANTS.MIME_CE_BATCH) as boolean; // A batch of events
 }
 
-/**
- * Creates a Kafka key for the event provided
- * @param {CloudEvent} e an event
- * @returns {string} the Kafka key
- */
-function createKey<T>(e: CloudEvent<T>): string {
-  return `${e.source}/${e.id}`;
-}
-
 /**
  * Determines what content mode a Kafka message is in given the provided headers
  * @param {Headers} headers the headers
@@ -175,7 +171,7 @@ function getMode(headers: Headers): Mode {
  * @param {KafkaMessage} message the message
  * @returns {CloudEvent<T>} a CloudEvent<T>
  */
-function parseBinary<T>(message: KafkaMessage<T>): KafkaEvent<T> {
+function parseBinary<T>(message: KafkaMessage<T>): CloudEvent<T> {
   const eventObj: { [key: string ]: unknown } = {};
   const headers = { ...message.headers };
 
@@ -196,7 +192,7 @@ function parseBinary<T>(message: KafkaMessage<T>): KafkaEvent<T> {
     }
   }
 
-  return new KafkaEvent({
+  return new CloudEvent({
     ...eventObj,
     data: extractBinaryData(message),
     partitionkey: message.key,
@@ -205,8 +201,8 @@ function parseBinary<T>(message: KafkaMessage<T>): KafkaEvent<T> {
 
 /**
  * Parses a structured kafka CE message and returns a CloudEvent
- * @param {KafkaMessage} message the message
- * @returns {CloudEvent<T>} a CloudEvent<T>
+ * @param {KafkaMessage<T>} message the message
+ * @returns {CloudEvent<T>} a KafkaEvent<T>
  */
 function parseStructured<T>(message: KafkaMessage<T>): CloudEvent<T> {
   // Although the format of a structured encoded event could be something
@@ -215,13 +211,17 @@ function parseStructured<T>(message: KafkaMessage<T>): CloudEvent<T> {
   if (!message.headers[CONSTANTS.HEADER_CONTENT_TYPE]?.startsWith(CONSTANTS.MIME_CE_JSON)) {
     throw new ValidationError(`Unsupported event encoding ${message.headers[CONSTANTS.HEADER_CONTENT_TYPE]}`);
   }
-  return new CloudEvent(JSON.parse(message.value as string), false);
+
+  return new CloudEvent({
+    ...JSON.parse(message.value as string),
+    partitionkey: message.key,
+  }, false);
 }
 
 /**
  * Parses a batch kafka CE message and returns a CloudEvent[]
- * @param {KafkaMessage} message the message
- * @returns {CloudEvent<T>} a CloudEvent<T>[]
+ * @param {KafkaMessage<T>} message the message
+ * @returns {CloudEvent<T>[]} an array of KafkaEvent<T>
  */
 function parseBatched<T>(message: KafkaMessage<T>): CloudEvent<T>[] {
   // Although the format of batch encoded events could be something
@@ -231,7 +231,7 @@ function parseBatched<T>(message: KafkaMessage<T>): CloudEvent<T>[] {
     throw new ValidationError(`Unsupported event encoding ${message.headers[CONSTANTS.HEADER_CONTENT_TYPE]}`);
   }
   const events = JSON.parse(message.value as string) as Record<string, unknown>[];
-  return events.map((e) => new CloudEvent(e, false));
+  return events.map((e) => new CloudEvent({ ...e, partitionkey: message.key }, false));
 }
 
 /**