You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/11/05 07:12:18 UTC
[GitHub] sijie closed pull request #2902: change kafkaConnectSource to
return KeyValue type record
sijie closed pull request #2902: change kafkaConnectSource to return KeyValue type record
URL: https://github.com/apache/pulsar/pull/2902
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-client-schema/src/main/java/org/apache/pulsar/client/api/Schema.java b/pulsar-client-schema/src/main/java/org/apache/pulsar/client/api/Schema.java
index e7f32b94dc..8797eaa020 100644
--- a/pulsar-client-schema/src/main/java/org/apache/pulsar/client/api/Schema.java
+++ b/pulsar-client-schema/src/main/java/org/apache/pulsar/client/api/Schema.java
@@ -122,6 +122,11 @@ default void validate(byte[] message) {
}
}
+ /**
+ * Schema that can be used to encode/decode KeyValue.
+ */
+ Schema<KeyValue<byte[], byte[]>> KV_BYTES = new KeyValueSchema(BYTES, BYTES);
+
/**
* Key Value Schema whose underneath key and value schemas are JSONSchema.
*/
diff --git a/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchema.java b/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchema.java
index d0cd796a09..0a62a6df5e 100644
--- a/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchema.java
+++ b/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchema.java
@@ -38,7 +38,7 @@
// schemaInfo combined by KeySchemaInfo and ValueSchemaInfo:
// [keyInfo.length][keyInfo][valueInfo.length][ValueInfo]
- private SchemaInfo schemaInfo;
+ private final SchemaInfo schemaInfo;
public KeyValueSchema(Schema<K> keySchema,
Schema<V> valueSchema) {
diff --git a/pulsar-client-schema/src/test/java/org/apache/pulsar/client/schema/KeyValueSchemaTest.java b/pulsar-client-schema/src/test/java/org/apache/pulsar/client/schema/KeyValueSchemaTest.java
index 04d0986f0b..23d502264f 100644
--- a/pulsar-client-schema/src/test/java/org/apache/pulsar/client/schema/KeyValueSchemaTest.java
+++ b/pulsar-client-schema/src/test/java/org/apache/pulsar/client/schema/KeyValueSchemaTest.java
@@ -118,4 +118,32 @@ public void testSchemaEncodeAndDecode() {
assertEquals(foo, fooBack);
assertEquals(bar, barBack);
}
+
+ @Test
+ public void testBytesSchemaEncodeAndDecode() {
+ AvroSchema<Foo> fooAvroSchema = AvroSchema.of(Foo.class);
+ AvroSchema<Bar> barAvroSchema = AvroSchema.of(Bar.class);
+
+ Bar bar = new Bar();
+ bar.setField1(true);
+
+ Foo foo = new Foo();
+ foo.setField1("field1");
+ foo.setField2("field2");
+ foo.setField3(3);
+ foo.setField4(bar);
+ foo.setColor(Color.RED);
+
+ byte[] fooBytes = fooAvroSchema.encode(foo);
+ byte[] barBytes = barAvroSchema.encode(bar);
+
+ byte[] encodeBytes = Schema.KV_BYTES.encode(new KeyValue<>(fooBytes, barBytes));
+ KeyValue<byte[], byte[]> decodeKV = Schema.KV_BYTES.decode(encodeBytes);
+
+ Foo fooBack = fooAvroSchema.decode(decodeKV.getKey());
+ Bar barBack = barAvroSchema.decode(decodeKV.getValue());
+
+ assertEquals(foo, fooBack);
+ assertEquals(bar, barBack);
+ }
}
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
index cbf97f1330..01ab9586f8 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
@@ -31,6 +31,7 @@
import org.apache.pulsar.client.impl.schema.AvroSchema;
import org.apache.pulsar.client.impl.schema.JSONSchema;
import org.apache.pulsar.client.impl.schema.ProtobufSchema;
+import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.functions.api.SerDe;
@@ -102,6 +103,8 @@ private static SchemaType getDefaultSchemaType(Class<?> clazz) {
return SchemaType.STRING;
} else if (isProtobufClass(clazz)) {
return SchemaType.PROTOBUF;
+ } else if (KeyValue.class.equals(clazz)) {
+ return SchemaType.KEY_VALUE;
} else {
return DEFAULT_SCHEMA_TYPE;
}
@@ -126,6 +129,9 @@ private static SchemaType getDefaultSchemaType(Class<?> clazz) {
case JSON:
return JSONSchema.of(clazz);
+ case KEY_VALUE:
+ return (Schema<T>)Schema.KV_BYTES;
+
case PROTOBUF:
return ProtobufSchema.ofGenericClass(clazz, Collections.emptyMap());
diff --git a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
index 5aac5ddf3c..a1647033a9 100644
--- a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
+++ b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
@@ -21,12 +21,16 @@
import java.util.Base64;
import java.util.Collections;
import java.util.HashMap;
-import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@@ -39,6 +43,7 @@
import org.apache.kafka.connect.storage.OffsetStorageReader;
import org.apache.kafka.connect.storage.OffsetStorageReaderImpl;
import org.apache.kafka.connect.storage.OffsetStorageWriter;
+import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.io.core.Source;
import org.apache.pulsar.io.core.SourceContext;
@@ -47,7 +52,7 @@
* A pulsar source that runs
*/
@Slf4j
-public class KafkaConnectSource implements Source<byte[]> {
+public class KafkaConnectSource implements Source<KeyValue<byte[], byte[]>> {
// kafka connect related variables
private SourceTaskContext sourceTaskContext;
@@ -63,7 +68,8 @@
private OffsetStorageReader offsetReader;
@Getter
private OffsetStorageWriter offsetWriter;
- private IdentityHashMap<SourceRecord, SourceRecord> outstandingRecords = new IdentityHashMap<>();
+ // number of outstandingRecords that have been polled but not been acked
+ private AtomicInteger outstandingRecords = new AtomicInteger(0);
@Override
public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception {
@@ -80,7 +86,6 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws
.getDeclaredConstructor()
.newInstance();
-
// initialize the key and value converter
keyConverter = ((Class<? extends Converter>)Class.forName(stringConfig.get(PulsarKafkaWorkerConfig.KEY_CONVERTER_CLASS_CONFIG)))
.asSubclass(Converter.class)
@@ -118,111 +123,146 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws
}
@Override
- public Record<byte[]> read() throws Exception {
+ public synchronized Record<KeyValue<byte[], byte[]>> read() throws Exception {
while (true) {
if (currentBatch == null) {
flushFuture = new CompletableFuture<>();
- List<SourceRecord> recordList = sourceTask.poll();
- if (recordList == null) {
+ List<SourceRecord> recordList = sourceTask.poll();
+ if (recordList == null || recordList.isEmpty()) {
Thread.sleep(1000);
continue;
}
+ outstandingRecords.addAndGet(recordList.size());
currentBatch = recordList.iterator();
}
if (currentBatch.hasNext()) {
return processSourceRecord(currentBatch.next());
} else {
- boolean hasOutstandingRecords;
- synchronized (this) {
- hasOutstandingRecords = !outstandingRecords.isEmpty();
- }
- if (!hasOutstandingRecords) {
- // there is no records any more, then waiting for the batch to complete writing
- // to sink and the offsets are committed as well
- flushFuture.get();
- flushFuture = null;
- }
+ // there is no records any more, then waiting for the batch to complete writing
+ // to sink and the offsets are committed as well, then do next round read.
+ flushFuture.get();
+ flushFuture = null;
currentBatch = null;
}
}
}
- private synchronized Record<byte[]> processSourceRecord(final SourceRecord srcRecord) {
- outstandingRecords.put(srcRecord, srcRecord);
+ @Override
+ public void close() {
+ sourceTask.stop();
+ }
+
+ private synchronized Record<KeyValue<byte[], byte[]>> processSourceRecord(final SourceRecord srcRecord) {
+ KafkaSourceRecord record = new KafkaSourceRecord(srcRecord);
offsetWriter.offset(srcRecord.sourcePartition(), srcRecord.sourceOffset());
- return new Record<byte[]>() {
- @Override
- public Optional<String> getKey() {
- byte[] keyBytes = keyConverter.fromConnectData(
- srcRecord.topic(), srcRecord.keySchema(), srcRecord.key());
- return Optional.of(Base64.getEncoder().encodeToString(keyBytes));
- }
+ return record;
+ }
- @Override
- public byte[] getValue() {
- return valueConverter.fromConnectData(
- srcRecord.topic(), srcRecord.valueSchema(), srcRecord.value());
- }
+ private static Map<String, String> PROPERTIES = Collections.emptyMap();
+ private static Optional<Long> RECORD_SEQUENCE = Optional.empty();
+ private static long FLUSH_TIMEOUT_MS = 2000;
- @Override
- public Optional<String> getTopicName() {
- return Optional.of(srcRecord.topic());
- }
+ private class KafkaSourceRecord implements Record<KeyValue<byte[], byte[]>> {
+ @Getter
+ Optional<String> key;
+ @Getter
+ KeyValue<byte[], byte[]> value;
+ @Getter
+ Optional<String> topicName;
+ @Getter
+ Optional<Long> eventTime;
+ @Getter
+ Optional<String> partitionId;
+ @Getter
+ Optional<String> destinationTopic;
- @Override
- public Optional<Long> getEventTime() {
- return Optional.ofNullable(srcRecord.timestamp());
+ KafkaSourceRecord(SourceRecord srcRecord) {
+ byte[] keyBytes = keyConverter.fromConnectData(
+ srcRecord.topic(), srcRecord.keySchema(), srcRecord.key());
+ byte[] valueBytes = valueConverter.fromConnectData(
+ srcRecord.topic(), srcRecord.valueSchema(), srcRecord.value());
+ if (keyBytes != null) {
+ this.key = Optional.of(Base64.getEncoder().encodeToString(keyBytes));
}
+ this.value = new KeyValue(keyBytes, valueBytes);
- @Override
- public Optional<String> getPartitionId() {
- String partitionId = srcRecord.sourcePartition()
- .entrySet()
- .stream()
- .map(e -> e.getKey() + "=" + e.getValue())
- .collect(Collectors.joining(","));
- return Optional.of(partitionId);
- }
+ this.topicName = Optional.of(srcRecord.topic());
+ this.eventTime = Optional.ofNullable(srcRecord.timestamp());
+ this.partitionId = Optional.of(srcRecord.sourcePartition()
+ .entrySet()
+ .stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .collect(Collectors.joining(",")));
+ this.destinationTopic = Optional.of(srcRecord.topic());
+ }
- @Override
- public Optional<Long> getRecordSequence() {
- return Optional.empty();
- }
+ @Override
+ public Optional<Long> getRecordSequence() {
+ return RECORD_SEQUENCE;
+ }
- @Override
- public Map<String, String> getProperties() {
- return Collections.emptyMap();
+ @Override
+ public Map<String, String> getProperties() {
+ return PROPERTIES;
+ }
+
+ private void completedFlushOffset(Throwable error, Void result) {
+ if (error != null) {
+ log.error("Failed to flush offsets to storage: ", error);
+ currentBatch = null;
+ offsetWriter.cancelFlush();
+ flushFuture.completeExceptionally(new Exception("No Offsets Added Error"));
+ } else {
+ log.trace("Finished flushing offsets to storage");
+ currentBatch = null;
+ flushFuture.complete(null);
}
+ }
+
+ @Override
+ public void ack() {
+ // TODO: should flush for each batch. not wait for a time for acked all.
+ // How to handle order between each batch. QueueList<pair<batch, automicInt>>. check if head is all acked.
+ boolean canFlush = (outstandingRecords.decrementAndGet() == 0);
- @Override
- public void ack() {
- boolean canComplete;
- synchronized (KafkaConnectSource.this) {
- outstandingRecords.remove(srcRecord);
- canComplete = outstandingRecords.isEmpty();
+ // consumed all the records, flush the offsets
+ if (canFlush && flushFuture != null) {
+ if (!offsetWriter.beginFlush()) {
+ log.error("When beginFlush, No offsets to commit!");
+ flushFuture.completeExceptionally(new Exception("No Offsets Added Error when beginFlush"));
+ return;
}
- if (canComplete && flushFuture != null) {
- flushFuture.complete(null);
+
+ Future<Void> doFlush = offsetWriter.doFlush(this::completedFlushOffset);
+ if (doFlush == null) {
+ // Offsets added in processSourceRecord, But here no offsets to commit
+ log.error("No offsets to commit!");
+ flushFuture.completeExceptionally(new Exception("No Offsets Added Error"));
+ return;
}
- }
- @Override
- public void fail() {
- if (flushFuture != null) {
- flushFuture.completeExceptionally(new Exception("Sink Error"));
+ // Wait until the offsets are flushed
+ try {
+ doFlush.get(FLUSH_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ sourceTask.commit();
+ } catch (InterruptedException e) {
+ log.warn("Flush of {} offsets interrupted, cancelling", this);
+ offsetWriter.cancelFlush();
+ } catch (ExecutionException e) {
+ log.error("Flush of {} offsets threw an unexpected exception: ", this, e);
+ offsetWriter.cancelFlush();
+ } catch (TimeoutException e) {
+ log.error("Timed out waiting to flush {} offsets to storage", this);
+ offsetWriter.cancelFlush();
}
}
+ }
- @Override
- public Optional<String> getDestinationTopic() {
- return Optional.of(srcRecord.topic());
+ @Override
+ public void fail() {
+ if (flushFuture != null) {
+ flushFuture.completeExceptionally(new Exception("Sink Error"));
}
- };
- }
-
-
- @Override
- public void close() throws Exception {
- sourceTask.stop();
+ }
}
}
diff --git a/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java b/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java
index e852c4c916..f887b3969f 100644
--- a/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java
+++ b/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java
@@ -18,20 +18,21 @@
*/
package org.apache.pulsar.io.kafka.connect;
-import static org.apache.kafka.connect.file.FileStreamSourceTask.FILENAME_FIELD;
-import static org.apache.kafka.connect.file.FileStreamSourceTask.POSITION_FIELD;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
import java.io.File;
import java.io.OutputStream;
+import java.nio.ByteBuffer;
import java.nio.file.Files;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.file.FileStreamSourceConnector;
import org.apache.kafka.connect.runtime.TaskConfig;
import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.functions.api.Record;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -77,36 +78,57 @@ protected void cleanup() throws Exception {
tempFile.delete();
super.internalCleanup();
}
+ protected void completedFlush(Throwable error, Void result) {
+ if (error != null) {
+ log.error("Failed to flush {} offsets to storage: ", this, error);
+ } else {
+ log.info("Finished flushing {} offsets to storage", this);
+ }
+ }
@Test
public void testOpenAndRead() throws Exception {
kafkaConnectSource = new KafkaConnectSource();
kafkaConnectSource.open(config, null);
+ // use FileStreamSourceConnector, each line is a record, need "\n" and end of each record.
OutputStream os = Files.newOutputStream(tempFile.toPath());
+
String line1 = "This is the first line\n";
os.write(line1.getBytes());
os.flush();
log.info("write 2 lines.");
- kafkaConnectSource.getOffsetWriter().offset(
- Collections.singletonMap(FILENAME_FIELD, config.get(FileStreamSourceConnector.FILE_CONFIG).toString()),
- Collections.singletonMap(POSITION_FIELD, /*tempFile.getTotalSpace()*/0L));
- // offset of second line
- long offset = tempFile.getTotalSpace();
String line2 = "This is the second line\n";
os.write(line2.getBytes());
os.flush();
- kafkaConnectSource.getOffsetWriter().offset(
- Collections.singletonMap(FILENAME_FIELD, config.get(FileStreamSourceConnector.FILE_CONFIG).toString()),
- Collections.singletonMap(POSITION_FIELD, offset));
log.info("finish write, will read 2 lines");
- Record<byte[]> record = kafkaConnectSource.read();
- assertTrue(line1.contains(new String(record.getValue())));
+ // Note: FileStreamSourceTask read the whole line as Value, and set Key as null.
+ Record<KeyValue<byte[], byte[]>> record = kafkaConnectSource.read();
+ String readBack1 = new String(record.getValue().getValue());
+ assertTrue(line1.contains(readBack1));
+ assertEquals(record.getValue().getKey(), null);
+ log.info("read line1: {}", readBack1);
+ record.ack();
+
+ record = kafkaConnectSource.read();
+ String readBack2 = new String(record.getValue().getValue());
+ assertTrue(line2.contains(readBack2));
+ assertEquals(record.getValue().getKey(), null);
+ log.info("read line2: {}", readBack2);
+ record.ack();
+
+ String line3 = "This is the 3rd line\n";
+ os.write(line3.getBytes());
+ os.flush();
record = kafkaConnectSource.read();
- assertTrue(line2.contains(new String(record.getValue())));
+ String readBack3 = new String(record.getValue().getValue());
+ assertTrue(line3.contains(readBack3));
+ assertEquals(record.getValue().getKey(), null);
+ log.info("read line3: {}", readBack3);
+ record.ack();
}
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services