eolivelli commented on a change in pull request #9448:
URL: https://github.com/apache/pulsar/pull/9448#discussion_r577402048
##########
File path:
pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSource.java
##########
@@ -43,14 +49,64 @@
@Override
protected Properties beforeCreateConsumer(Properties props) {
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+ props.putIfAbsent("schema.registry.url", "http://localhost:8081";);
+ props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+
+ String currentValue =
props.getProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);
+
+ // replace KafkaAvroDeserializer with our custom implementation
+ if (currentValue != null &&
currentValue.equals(KafkaAvroDeserializer.class.getName())) {
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
NoCopyKafkaAvroDeserializer.class.getName());
+ }
+
log.info("Created kafka consumer config : {}", props);
return props;
}
@Override
- public byte[] extractValue(ConsumerRecord<String, byte[]> record) {
- return record.value();
+ public Object extractValue(ConsumerRecord<String, Object> consumerRecord) {
+ Object value = consumerRecord.value();
+ if (value instanceof BytesWithAvroPulsarSchema) {
+ return ((BytesWithAvroPulsarSchema) value).getValue();
+ }
+ return value;
+ }
+
+ @Override
+ public org.apache.pulsar.client.api.Schema<?>
extractSchema(ConsumerRecord<String, Object> consumerRecord) {
+ Object value = consumerRecord.value();
+ if (value instanceof BytesWithAvroPulsarSchema) {
+ return ((BytesWithAvroPulsarSchema) value).getPulsarSchema();
+ } else {
+ return org.apache.pulsar.client.api.Schema.BYTES;
+ }
+ }
+
+ public static class NoCopyKafkaAvroDeserializer extends
KafkaAvroDeserializer {
+
+ private final PulsarSchemaCache<GenericRecord> schemaCache = new
PulsarSchemaCache<>();
+
+ @Override
+ protected Object deserialize(boolean includeSchemaAndVersion, String
topic, Boolean isKey, byte[] payload, Schema readerSchema) throws
SerializationException {
+ if (payload == null) {
+ return null;
+ } else {
+ int id = -1;
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(payload);
+ buffer.get(); // magic number
+ id = buffer.getInt();
+ String subject = getSubjectName(topic, isKey != null ?
isKey : false);
+ Schema schema =
this.schemaRegistry.getBySubjectAndId(subject, id);
Review comment:
The `schemaRegistry` has its own cache (it should be a
`CachedSchemaRegistryClient`)
This version of the code is a simplified version of the original
`deserialize` method.
I wanted to keep it as close as possible to the original implementation.
```
protected Object deserialize(boolean includeSchemaAndVersion, String topic,
Boolean isKey, byte[] payload, Schema readerSchema) throws
SerializationException {
if (payload == null) {
return null;
} else {
byte id = -1;
try {
ByteBuffer buffer = this.getByteBuffer(payload);
int id = buffer.getInt();
String subject = includeSchemaAndVersion ?
getSubjectName(topic, isKey) : null;
Schema schema =
this.schemaRegistry.getBySubjectAndId(subject, id);
int length = buffer.limit() - 1 - 4;
Object result;
```
##########
File path:
pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSource.java
##########
@@ -43,14 +49,64 @@
@Override
protected Properties beforeCreateConsumer(Properties props) {
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+ props.putIfAbsent("schema.registry.url", "http://localhost:8081";);
+ props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+
+ String currentValue =
props.getProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);
+
+ // replace KafkaAvroDeserializer with our custom implementation
+ if (currentValue != null &&
currentValue.equals(KafkaAvroDeserializer.class.getName())) {
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
NoCopyKafkaAvroDeserializer.class.getName());
+ }
+
Review comment:
sure. I added a validation
##########
File path:
pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSource.java
##########
@@ -43,14 +49,64 @@
@Override
protected Properties beforeCreateConsumer(Properties props) {
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+ props.putIfAbsent("schema.registry.url", "http://localhost:8081";);
+ props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+
+ String currentValue =
props.getProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);
+
+ // replace KafkaAvroDeserializer with our custom implementation
+ if (currentValue != null &&
currentValue.equals(KafkaAvroDeserializer.class.getName())) {
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
NoCopyKafkaAvroDeserializer.class.getName());
+ }
+
log.info("Created kafka consumer config : {}", props);
return props;
}
@Override
- public byte[] extractValue(ConsumerRecord<String, byte[]> record) {
- return record.value();
+ public Object extractValue(ConsumerRecord<String, Object> consumerRecord) {
+ Object value = consumerRecord.value();
+ if (value instanceof BytesWithAvroPulsarSchema) {
+ return ((BytesWithAvroPulsarSchema) value).getValue();
+ }
+ return value;
+ }
+
+ @Override
+ public org.apache.pulsar.client.api.Schema<?>
extractSchema(ConsumerRecord<String, Object> consumerRecord) {
+ Object value = consumerRecord.value();
+ if (value instanceof BytesWithAvroPulsarSchema) {
+ return ((BytesWithAvroPulsarSchema) value).getPulsarSchema();
+ } else {
+ return org.apache.pulsar.client.api.Schema.BYTES;
+ }
+ }
+
+ public static class NoCopyKafkaAvroDeserializer extends
KafkaAvroDeserializer {
+
+ private final PulsarSchemaCache<GenericRecord> schemaCache = new
PulsarSchemaCache<>();
+
+ @Override
+ protected Object deserialize(boolean includeSchemaAndVersion, String
topic, Boolean isKey, byte[] payload, Schema readerSchema) throws
SerializationException {
+ if (payload == null) {
+ return null;
+ } else {
+ int id = -1;
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(payload);
+ buffer.get(); // magic number
+ id = buffer.getInt();
+ String subject = getSubjectName(topic, isKey != null ?
isKey : false);
+ Schema schema =
this.schemaRegistry.getBySubjectAndId(subject, id);
Review comment:
In the end we have to pass a byte[] to the PulsarProducer as value for
the Record (we are implementing a `Source<byte[]>`, so we have to copy this
data at least once.
There is no way to save this memory copy, because we have to skip the magic
number and the schema id.
##########
File path:
pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/PulsarSchemaCache.java
##########
@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.pulsar.io.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.common.schema.SchemaType;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.IdentityHashMap;
+
+@Slf4j
+class PulsarSchemaCache<T> {
Review comment:
Updated
##########
File path:
pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSource.java
##########
@@ -43,14 +49,64 @@
@Override
protected Properties beforeCreateConsumer(Properties props) {
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+ props.putIfAbsent("schema.registry.url", "http://localhost:8081";);
+ props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+
+ String currentValue =
props.getProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);
+
+ // replace KafkaAvroDeserializer with our custom implementation
+ if (currentValue != null &&
currentValue.equals(KafkaAvroDeserializer.class.getName())) {
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
NoCopyKafkaAvroDeserializer.class.getName());
+ }
+
log.info("Created kafka consumer config : {}", props);
return props;
}
@Override
- public byte[] extractValue(ConsumerRecord<String, byte[]> record) {
- return record.value();
+ public Object extractValue(ConsumerRecord<String, Object> consumerRecord) {
Review comment:
makes sense.
I am adding a new `extractKey` method in order to make it clearer
##########
File path:
pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSource.java
##########
@@ -43,14 +49,64 @@
@Override
protected Properties beforeCreateConsumer(Properties props) {
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+ props.putIfAbsent("schema.registry.url", "http://localhost:8081";);
Review comment:
removed (it was just an example value, useful only for local testing)
##########
File path:
pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSource.java
##########
@@ -43,14 +49,64 @@
@Override
protected Properties beforeCreateConsumer(Properties props) {
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+ props.putIfAbsent("schema.registry.url", "http://localhost:8081";);
+ props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+
+ String currentValue =
props.getProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);
+
+ // replace KafkaAvroDeserializer with our custom implementation
+ if (currentValue != null &&
currentValue.equals(KafkaAvroDeserializer.class.getName())) {
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
NoCopyKafkaAvroDeserializer.class.getName());
+ }
+
log.info("Created kafka consumer config : {}", props);
return props;
}
@Override
- public byte[] extractValue(ConsumerRecord<String, byte[]> record) {
- return record.value();
+ public Object extractValue(ConsumerRecord<String, Object> consumerRecord) {
+ Object value = consumerRecord.value();
+ if (value instanceof BytesWithAvroPulsarSchema) {
+ return ((BytesWithAvroPulsarSchema) value).getValue();
+ }
+ return value;
+ }
+
+ @Override
+ public org.apache.pulsar.client.api.Schema<?>
extractSchema(ConsumerRecord<String, Object> consumerRecord) {
+ Object value = consumerRecord.value();
+ if (value instanceof BytesWithAvroPulsarSchema) {
+ return ((BytesWithAvroPulsarSchema) value).getPulsarSchema();
Review comment:
unfortunately I don't have a way to pass the PulsarAvroSchemaCache
instance to the deserializer (because we do not have control over the
lifecycle).
Also we also need one instance and its lifecycle is tied to the lifecycle of
the deserializer, so IMHO it can live internally to the deserializer.
Actually this deserializer produces `BytesWithAvroPulsarSchema` instances. I
believe it is fine to leave all of the schema management stuff into it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
