String = 0;
+ */
+ String(0),
+ /**
+ * Integer = 1;
+ */
+ Integer(1),
+ /**
+ * Double = 2;
+ */
+ Double(2),
+ /**
+ * ByteArray = 3;
+ */
+ ByteArray(3),
+ UNRECOGNIZED(-1),
+ ;
+
+ /**
+ * String = 0;
+ */
+ public static final int String_VALUE = 0;
+ /**
+ * Integer = 1;
+ */
+ public static final int Integer_VALUE = 1;
+ /**
+ * Double = 2;
+ */
+ public static final int Double_VALUE = 2;
+ /**
+ * ByteArray = 3;
+ */
+ public static final int ByteArray_VALUE = 3;
+
+
+ public final int getNumber() {
+ if (this == UNRECOGNIZED) {
+ throw new java.lang.IllegalArgumentException(
+ "Can't get the number of an unknown enum value.");
+ }
+ return value;
+ }
+
+ /**
+ * @param value The numeric wire value of the corresponding enum entry.
+ * @return The enum associated with the given numeric wire value.
+ * @deprecated Use {@link #forNumber(int)} instead.
+ */
+ @java.lang.Deprecated
+ public static KeyType valueOf(int value) {
+ return forNumber(value);
+ }
+
+ /**
+ * @param value The numeric wire value of the corresponding enum entry.
+ * @return The enum associated with the given numeric wire value.
+ */
+ public static KeyType forNumber(int value) {
+ switch (value) {
+ case 0: return String;
+ case 1: return Integer;
+ case 2: return Double;
+ case 3: return ByteArray;
+ default: return null;
+ }
+ }
+
+ public static com.google.protobuf.Internal.EnumLiteMap
* CloudEvent content mode
@@ -340,7 +466,7 @@ public ContentMode findValueByNumber(int number) {
}
public static final com.google.protobuf.Descriptors.EnumDescriptor
getDescriptor() {
- return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.getDescriptor().getEnumTypes().get(2);
+ return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.getDescriptor().getEnumTypes().get(3);
}
private static final ContentMode[] VALUES = values();
@@ -2910,6 +3036,25 @@ public interface EgressOrBuilder extends
*/
dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DeliveryOrder getDeliveryOrder();
+ /**
+ *
+ * Kafka record key type.
+ *
+ *
+ * .KeyType keyType = 10;
+ * @return The enum numeric value on the wire for keyType.
+ */
+ int getKeyTypeValue();
+ /**
+ *
+ * Kafka record key type.
+ *
+ *
+ * .KeyType keyType = 10;
+ * @return The keyType.
+ */
+ dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType getKeyType();
+
public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress.ReplyStrategyCase getReplyStrategyCase();
}
/**
@@ -2929,6 +3074,7 @@ private Egress() {
destination_ = "";
uid_ = "";
deliveryOrder_ = 0;
+ keyType_ = 0;
}
@java.lang.Override
@@ -3045,6 +3191,12 @@ private Egress(
replyStrategyCase_ = 9;
break;
}
+ case 80: {
+ int rawValue = input.readEnum();
+
+ keyType_ = rawValue;
+ break;
+ }
default: {
if (!parseUnknownField(
input, unknownFields, extensionRegistry, tag)) {
@@ -3506,6 +3658,33 @@ public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressConfig
return result == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DeliveryOrder.UNRECOGNIZED : result;
}
+ public static final int KEYTYPE_FIELD_NUMBER = 10;
+ private int keyType_;
+ /**
+ *
+ * Kafka record key type.
+ *
+ *
+ * .KeyType keyType = 10;
+ * @return The enum numeric value on the wire for keyType.
+ */
+ @java.lang.Override public int getKeyTypeValue() {
+ return keyType_;
+ }
+ /**
+ *
+ * Kafka record key type.
+ *
+ *
+ * .KeyType keyType = 10;
+ * @return The keyType.
+ */
+ @java.lang.Override public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType getKeyType() {
+ @SuppressWarnings("deprecation")
+ dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType result = dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType.valueOf(keyType_);
+ return result == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType.UNRECOGNIZED : result;
+ }
+
private byte memoizedIsInitialized = -1;
@java.lang.Override
public final boolean isInitialized() {
@@ -3547,6 +3726,9 @@ public void writeTo(com.google.protobuf.CodedOutputStream output)
if (replyStrategyCase_ == 9) {
output.writeMessage(9, (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_);
}
+ if (keyType_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType.String.getNumber()) {
+ output.writeEnum(10, keyType_);
+ }
unknownFields.writeTo(output);
}
@@ -3588,6 +3770,10 @@ public int getSerializedSize() {
size += com.google.protobuf.CodedOutputStream
.computeMessageSize(9, (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_);
}
+ if (keyType_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType.String.getNumber()) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeEnumSize(10, keyType_);
+ }
size += unknownFields.getSerializedSize();
memoizedSize = size;
return size;
@@ -3620,6 +3806,7 @@ public boolean equals(final java.lang.Object obj) {
.equals(other.getEgressConfig())) return false;
}
if (deliveryOrder_ != other.deliveryOrder_) return false;
+ if (keyType_ != other.keyType_) return false;
if (!getReplyStrategyCase().equals(other.getReplyStrategyCase())) return false;
switch (replyStrategyCase_) {
case 3:
@@ -3664,6 +3851,8 @@ public int hashCode() {
}
hash = (37 * hash) + DELIVERYORDER_FIELD_NUMBER;
hash = (53 * hash) + deliveryOrder_;
+ hash = (37 * hash) + KEYTYPE_FIELD_NUMBER;
+ hash = (53 * hash) + keyType_;
switch (replyStrategyCase_) {
case 3:
hash = (37 * hash) + REPLYURL_FIELD_NUMBER;
@@ -3833,6 +4022,8 @@ public Builder clear() {
}
deliveryOrder_ = 0;
+ keyType_ = 0;
+
replyStrategyCase_ = 0;
replyStrategy_ = null;
return this;
@@ -3892,6 +4083,7 @@ public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress build
result.egressConfig_ = egressConfigBuilder_.build();
}
result.deliveryOrder_ = deliveryOrder_;
+ result.keyType_ = keyType_;
result.replyStrategyCase_ = replyStrategyCase_;
onBuilt();
return result;
@@ -3962,6 +4154,9 @@ public Builder mergeFrom(dev.knative.eventing.kafka.broker.contract.DataPlaneCon
if (other.deliveryOrder_ != 0) {
setDeliveryOrderValue(other.getDeliveryOrderValue());
}
+ if (other.keyType_ != 0) {
+ setKeyTypeValue(other.getKeyTypeValue());
+ }
switch (other.getReplyStrategyCase()) {
case REPLYURL: {
replyStrategyCase_ = 3;
@@ -5154,6 +5349,80 @@ public Builder clearDeliveryOrder() {
onChanged();
return this;
}
+
+ private int keyType_ = 0;
+ /**
+ *
+ * Kafka record key type.
+ *
+ *
+ * .KeyType keyType = 10;
+ * @return The enum numeric value on the wire for keyType.
+ */
+ @java.lang.Override public int getKeyTypeValue() {
+ return keyType_;
+ }
+ /**
+ *
+ * Kafka record key type.
+ *
+ *
+ * .KeyType keyType = 10;
+ * @param value The enum numeric value on the wire for keyType to set.
+ * @return This builder for chaining.
+ */
+ public Builder setKeyTypeValue(int value) {
+
+ keyType_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ *
+ * Kafka record key type.
+ *
+ *
+ * .KeyType keyType = 10;
+ * @return The keyType.
+ */
+ @java.lang.Override
+ public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType getKeyType() {
+ @SuppressWarnings("deprecation")
+ dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType result = dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType.valueOf(keyType_);
+ return result == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType.UNRECOGNIZED : result;
+ }
+ /**
+ *
+ * Kafka record key type.
+ *
+ *
+ * .KeyType keyType = 10;
+ * @param value The keyType to set.
+ * @return This builder for chaining.
+ */
+ public Builder setKeyType(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+
+ keyType_ = value.getNumber();
+ onChanged();
+ return this;
+ }
+ /**
+ *
+ * Kafka record key type.
+ *
+ *
+ * .KeyType keyType = 10;
+ * @return This builder for chaining.
+ */
+ public Builder clearKeyType() {
+
+ keyType_ = 0;
+ onChanged();
+ return this;
+ }
@java.lang.Override
public final Builder setUnknownFields(
final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -11657,31 +11926,34 @@ public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Contract get
"e\030\002 \001(\t:\0028\001\"\177\n\014EgressConfig\022\022\n\ndeadLette" +
"r\030\001 \001(\t\022\r\n\005retry\030\002 \001(\r\022%\n\rbackoffPolicy\030" +
"\003 \001(\0162\016.BackoffPolicy\022\024\n\014backoffDelay\030\004 " +
- "\001(\004\022\017\n\007timeout\030\005 \001(\004\"\223\002\n\006Egress\022\025\n\rconsu" +
+ "\001(\004\022\017\n\007timeout\030\005 \001(\004\"\256\002\n\006Egress\022\025\n\rconsu" +
"merGroup\030\001 \001(\t\022\023\n\013destination\030\002 \001(\t\022\022\n\010r" +
"eplyUrl\030\003 \001(\tH\000\022&\n\024replyToOriginalTopic\030" +
"\004 \001(\0132\006.EmptyH\000\022\036\n\014discardReply\030\t \001(\0132\006." +
"EmptyH\000\022\027\n\006filter\030\005 \001(\0132\007.Filter\022\013\n\003uid\030" +
"\006 \001(\t\022#\n\014egressConfig\030\007 \001(\0132\r.EgressConf" +
"ig\022%\n\rdeliveryOrder\030\010 \001(\0162\016.DeliveryOrde" +
- "rB\017\n\rreplyStrategy\"[\n\007Ingress\022!\n\013content" +
- "Mode\030\001 \001(\0162\014.ContentMode\022\016\n\004path\030\002 \001(\tH\000" +
- "\022\016\n\004host\030\003 \001(\tH\000B\r\n\013ingressType\"K\n\tRefer" +
- "ence\022\014\n\004uuid\030\001 \001(\t\022\021\n\tnamespace\030\002 \001(\t\022\014\n" +
- "\004name\030\003 \001(\t\022\017\n\007version\030\004 \001(\t\"\344\001\n\010Resourc" +
- "e\022\013\n\003uid\030\001 \001(\t\022\016\n\006topics\030\002 \003(\t\022\030\n\020bootst" +
- "rapServers\030\003 \001(\t\022\031\n\007ingress\030\004 \001(\0132\010.Ingr" +
- "ess\022#\n\014egressConfig\030\005 \001(\0132\r.EgressConfig" +
- "\022\031\n\010egresses\030\006 \003(\0132\007.Egress\022\034\n\nabsentAut" +
- "h\030\007 \001(\0132\006.EmptyH\000\022 \n\nauthSecret\030\010 \001(\0132\n." +
- "ReferenceH\000B\006\n\004Auth\"<\n\010Contract\022\022\n\ngener" +
- "ation\030\001 \001(\004\022\034\n\tresources\030\002 \003(\0132\t.Resourc" +
- "e*,\n\rBackoffPolicy\022\017\n\013Exponential\020\000\022\n\n\006L" +
- "inear\020\001*+\n\rDeliveryOrder\022\r\n\tUNORDERED\020\000\022" +
- "\013\n\007ORDERED\020\001*)\n\013ContentMode\022\n\n\006BINARY\020\000\022" +
- "\016\n\nSTRUCTURED\020\001B[\n*dev.knative.eventing." +
- "kafka.broker.contractB\021DataPlaneContract" +
- "Z\032control-plane/pkg/contractb\006proto3"
+ "r\022\031\n\007keyType\030\n \001(\0162\010.KeyTypeB\017\n\rreplyStr" +
+ "ategy\"[\n\007Ingress\022!\n\013contentMode\030\001 \001(\0162\014." +
+ "ContentMode\022\016\n\004path\030\002 \001(\tH\000\022\016\n\004host\030\003 \001(" +
+ "\tH\000B\r\n\013ingressType\"K\n\tReference\022\014\n\004uuid\030" +
+ "\001 \001(\t\022\021\n\tnamespace\030\002 \001(\t\022\014\n\004name\030\003 \001(\t\022\017" +
+ "\n\007version\030\004 \001(\t\"\344\001\n\010Resource\022\013\n\003uid\030\001 \001(" +
+ "\t\022\016\n\006topics\030\002 \003(\t\022\030\n\020bootstrapServers\030\003 " +
+ "\001(\t\022\031\n\007ingress\030\004 \001(\0132\010.Ingress\022#\n\014egress" +
+ "Config\030\005 \001(\0132\r.EgressConfig\022\031\n\010egresses\030" +
+ "\006 \003(\0132\007.Egress\022\034\n\nabsentAuth\030\007 \001(\0132\006.Emp" +
+ "tyH\000\022 \n\nauthSecret\030\010 \001(\0132\n.ReferenceH\000B\006" +
+ "\n\004Auth\"<\n\010Contract\022\022\n\ngeneration\030\001 \001(\004\022\034" +
+ "\n\tresources\030\002 \003(\0132\t.Resource*,\n\rBackoffP" +
+ "olicy\022\017\n\013Exponential\020\000\022\n\n\006Linear\020\001*+\n\rDe" +
+ "liveryOrder\022\r\n\tUNORDERED\020\000\022\013\n\007ORDERED\020\001*" +
+ "=\n\007KeyType\022\n\n\006String\020\000\022\013\n\007Integer\020\001\022\n\n\006D" +
+ "ouble\020\002\022\r\n\tByteArray\020\003*)\n\013ContentMode\022\n\n" +
+ "\006BINARY\020\000\022\016\n\nSTRUCTURED\020\001B[\n*dev.knative" +
+ ".eventing.kafka.broker.contractB\021DataPla" +
+ "neContractZ\032control-plane/pkg/contractb\006" +
+ "proto3"
};
descriptor = com.google.protobuf.Descriptors.FileDescriptor
.internalBuildGeneratedFileFrom(descriptorData,
@@ -11716,7 +11988,7 @@ public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Contract get
internal_static_Egress_fieldAccessorTable = new
com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
internal_static_Egress_descriptor,
- new java.lang.String[] { "ConsumerGroup", "Destination", "ReplyUrl", "ReplyToOriginalTopic", "DiscardReply", "Filter", "Uid", "EgressConfig", "DeliveryOrder", "ReplyStrategy", });
+ new java.lang.String[] { "ConsumerGroup", "Destination", "ReplyUrl", "ReplyToOriginalTopic", "DiscardReply", "Filter", "Uid", "EgressConfig", "DeliveryOrder", "KeyType", "ReplyStrategy", });
internal_static_Ingress_descriptor =
getDescriptor().getMessageTypes().get(4);
internal_static_Ingress_fieldAccessorTable = new
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
index bc83014429..86c20d33bb 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
@@ -22,6 +22,7 @@
import dev.knative.eventing.kafka.broker.core.utils.CollectionsUtils;
import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
+
import java.util.AbstractMap.SimpleImmutableEntry;
import java.util.ArrayList;
import java.util.Collection;
@@ -31,6 +32,7 @@
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Collectors;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -243,7 +245,8 @@ private boolean egressEquals(DataPlaneContract.Egress e1, DataPlaneContract.Egre
&& Objects.equals(e1.getReplyUrl(), e2.getReplyUrl())
&& Objects.equals(e1.getReplyToOriginalTopic(), e2.getReplyToOriginalTopic())
&& Objects.equals(e1.getEgressConfig(), e2.getEgressConfig())
- && Objects.equals(e1.getFilter(), e2.getFilter());
+ && Objects.equals(e1.getFilter(), e2.getFilter())
+ && Objects.equals(e1.getKeyType(), e2.getKeyType());
}
private static void logFailure(final String msg, final DataPlaneContract.Egress egress, final Throwable cause) {
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java
index c80eed25d4..b264922e3c 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java
@@ -32,6 +32,6 @@ public interface RecordDispatcher extends AsyncCloseable {
* @param record record to handle.
* @return the completion future.
*/
- Future dispatch(KafkaConsumerRecord record);
+ Future dispatch(KafkaConsumerRecord record);
}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
index 981121db35..0896d038fc 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
@@ -46,8 +46,8 @@ public class RecordDispatcherImpl implements RecordDispatcher {
private static final Logger logger = LoggerFactory.getLogger(RecordDispatcherImpl.class);
private final Filter filter;
- private final Function, Future> subscriberSender;
- private final Function, Future> dlsSender;
+ private final Function, Future> subscriberSender;
+ private final Function, Future> dlsSender;
private final RecordDispatcherListener recordDispatcherListener;
private final AsyncCloseable closeable;
private final ConsumerTracer consumerTracer;
@@ -89,7 +89,7 @@ public RecordDispatcherImpl(
* @param record record to handle.
*/
@Override
- public Future dispatch(KafkaConsumerRecord record) {
+ public Future dispatch(KafkaConsumerRecord record) {
Promise promise = Promise.promise();
/*
@@ -119,7 +119,7 @@ public Future dispatch(KafkaConsumerRecord record) {
return promise.future();
}
- private void onRecordReceived(final KafkaConsumerRecord record, Promise finalProm) {
+ private void onRecordReceived(final KafkaConsumerRecord record, Promise finalProm) {
logDebug("Handling record", record);
// Trace record received event
@@ -151,35 +151,35 @@ private void onRecordReceived(final KafkaConsumerRecord reco
.onFailure(finalProm::fail); // This should really never happen
}
- private void onFilterMatching(final KafkaConsumerRecord record, final Promise finalProm) {
+ private void onFilterMatching(final KafkaConsumerRecord record, final Promise finalProm) {
logDebug("Record match filtering", record);
subscriberSender.apply(record)
.onSuccess(res -> onSubscriberSuccess(record, finalProm))
.onFailure(ex -> onSubscriberFailure(record, finalProm));
}
- private void onFilterNotMatching(final KafkaConsumerRecord record,
+ private void onFilterNotMatching(final KafkaConsumerRecord record,
final Promise finalProm) {
logDebug("Record doesn't match filtering", record);
recordDispatcherListener.recordDiscarded(record)
.onComplete(finalProm);
}
- private void onSubscriberSuccess(final KafkaConsumerRecord record,
+ private void onSubscriberSuccess(final KafkaConsumerRecord record,
final Promise finalProm) {
logDebug("Successfully sent event to subscriber", record);
recordDispatcherListener.successfullySentToSubscriber(record)
.onComplete(finalProm);
}
- private void onSubscriberFailure(final KafkaConsumerRecord record,
+ private void onSubscriberFailure(final KafkaConsumerRecord record,
final Promise finalProm) {
dlsSender.apply(record)
.onSuccess(v -> onDeadLetterSinkSuccess(record, finalProm))
.onFailure(ex -> onDeadLetterSinkFailure(record, ex, finalProm));
}
- private void onDeadLetterSinkSuccess(final KafkaConsumerRecord record,
+ private void onDeadLetterSinkSuccess(final KafkaConsumerRecord record,
final Promise finalProm) {
logDebug("Successfully sent event to the dead letter sink", record);
recordDispatcherListener.successfullySentToDeadLetterSink(record)
@@ -187,13 +187,13 @@ private void onDeadLetterSinkSuccess(final KafkaConsumerRecord record, final Throwable exception,
+ private void onDeadLetterSinkFailure(final KafkaConsumerRecord record, final Throwable exception,
final Promise finalProm) {
recordDispatcherListener.failedToSendToDeadLetterSink(record, exception)
.onComplete(finalProm);
}
- private static Function, Future> composeSenderAndSinkHandler(
+ private static Function, Future> composeSenderAndSinkHandler(
CloudEventSender sender, ResponseHandler sinkHandler, String senderType) {
return rec -> sender.send(rec.value())
.onFailure(ex -> logError("Failed to send event to " + senderType, rec, ex))
@@ -205,7 +205,7 @@ private static Function, Future> c
private static void logError(
final String msg,
- final KafkaConsumerRecord record,
+ final KafkaConsumerRecord record,
final Throwable cause) {
if (logger.isDebugEnabled()) {
@@ -229,13 +229,14 @@ private static void logError(
private static void logDebug(
final String msg,
- final KafkaConsumerRecord record) {
+ final KafkaConsumerRecord record) {
- logger.debug(msg + " {} {} {} {} {}",
+ logger.debug(msg + " {} {} {} {} {} {}",
keyValue("topic", record.topic()),
keyValue("partition", record.partition()),
keyValue("headers", record.headers()),
keyValue("offset", record.offset()),
+ keyValue("key", record.key()),
keyValue("event", record.value())
);
}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
index d8f7754ffb..12fac0eb29 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
@@ -39,7 +39,7 @@ public interface Initializer extends BiFunction topics;
- KafkaConsumer consumer;
+ KafkaConsumer consumer;
RecordDispatcher recordDispatcher;
private AsyncCloseable closeable;
@@ -74,7 +74,7 @@ public void stop(Promise stopPromise) {
).close(stopPromise);
}
- public void setConsumer(KafkaConsumer consumer) {
+ public void setConsumer(KafkaConsumer consumer) {
this.consumer = consumer;
}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/InvalidCloudEventInterceptor.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/InvalidCloudEventInterceptor.java
index 12d1a1a396..88bc903b1d 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/InvalidCloudEventInterceptor.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/InvalidCloudEventInterceptor.java
@@ -18,7 +18,6 @@
import io.cloudevents.CloudEvent;
import io.cloudevents.core.v1.CloudEventBuilder;
-import io.cloudevents.kafka.PartitionKeyExtensionInterceptor;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -39,6 +38,7 @@
import java.util.stream.Stream;
import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
+import static io.cloudevents.kafka.PartitionKeyExtensionInterceptor.PARTITION_KEY_EXTENSION;
/**
* The {@link InvalidCloudEventInterceptor} is a {@link ConsumerInterceptor}.
@@ -51,7 +51,7 @@
*
* For this reason this interceptor is capable of creating a CloudEvent from {@link ConsumerRecord} metadata and data.
*/
-public class InvalidCloudEventInterceptor implements ConsumerInterceptor {
+public class InvalidCloudEventInterceptor implements ConsumerInterceptor {
private static final Logger logger = LoggerFactory.getLogger(InvalidCloudEventInterceptor.class);
@@ -61,6 +61,8 @@ public class InvalidCloudEventInterceptor implements ConsumerInterceptor configs) {
}
@Override
- public ConsumerRecords onConsume(final ConsumerRecords records) {
+ public ConsumerRecords onConsume(final ConsumerRecords records) {
if (!this.isEnabled || records == null || records.isEmpty()) {
return records;
}
- final Map>> validRecords = new HashMap<>(records.count());
+ final Map>> validRecords = new HashMap<>(records.count());
for (final var record : records) {
final var tp = new TopicPartition(record.topic(), record.partition());
if (!validRecords.containsKey(tp)) {
@@ -128,7 +130,7 @@ private static String getOrDefault(final Map configs, final String ke
return null;
}
- private ConsumerRecord validRecord(final ConsumerRecord record) {
+ private ConsumerRecord validRecord(final ConsumerRecord record) {
if (!(record.value() instanceof InvalidCloudEvent)) {
return record; // Valid CloudEvent
}
@@ -141,8 +143,9 @@ private ConsumerRecord validRecord(final ConsumerRecord validRecord(final ConsumerRecord= 'a' && c <= 'z') || (c >= '0' && c <= '9');
}
- private static String subject(final ConsumerRecord record) {
+ private static String subject(final ConsumerRecord record) {
return "partition:" + record.partition() + "#" + record.offset();
}
- private URI source(final ConsumerRecord record) {
+ private URI source(final ConsumerRecord record) {
return URI.create(
"/apis/v1/namespaces/" +
this.sourceNamespace +
@@ -214,11 +235,11 @@ private static String type() {
return TYPE;
}
- private static OffsetDateTime time(final ConsumerRecord record) {
+ private static OffsetDateTime time(final ConsumerRecord record) {
return OffsetDateTime.ofInstant(Instant.ofEpochMilli(record.timestamp()), ZoneId.of("UTC"));
}
- private static String id(final ConsumerRecord record) {
+ private static String id(final ConsumerRecord record) {
return "partition:" + record.partition() + "/offset:" + record.offset();
}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/KeyDeserializer.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/KeyDeserializer.java
new file mode 100644
index 0000000000..f8502e5d4e
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/KeyDeserializer.java
@@ -0,0 +1,107 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
+
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.DoubleDeserializer;
+import org.apache.kafka.common.serialization.FloatDeserializer;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+import java.util.Map;
+
+/**
+ * Deserializer for the key of Kafka records.
+ *
+ * It uses the configuration value associated with {@link KeyDeserializer.KEY_TYPE} to deserialize the key of a consumer record.
+ *
+ * Internally, it uses Kafka client provided deserializers for deserialization.
+ *
+ * - When {@code KEY_TYPE = String} it uses {@link StringDeserializer}.
+ *
+ * - When {@code KEY_TYPE = Integer} it uses {@link IntegerDeserializer} and falls back to {@link StringDeserializer}
+ * if the {@code byte[]} is not 4 bytes long.
+ *
+ * - When {@code KEY_TYPE = Double} it uses {@link FloatDeserializer} if the key is 4 bytes long,
+ * {@link DoubleDeserializer} if the key is 8 bytes long and falls back to {@link StringDeserializer} in any other case.
+ *
+ * - When {@code KEY_TYPE = ByteArray} it just returns the key.
+ *
+ * - When {@code KEY_TYPE = UNRECOGNIZED or unspecified} it uses {@link StringDeserializer}.
+ */
+public class KeyDeserializer implements Deserializer