sijie

#!/usr/bin/env bash
#
# 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
#

sijie

protected void testKafkaSourcePulsarFunction(TopicName pulsarTopic) throws Exception {
    final String tenantName = pulsarTopic.getTenant();
    final String namespaceName = pulsarTopic.getNamespacePortion();

    // create kafka topic
    final String kafkaTopic = "test-kafka-source-pulsar-function-" + Base58.randomString(8);
    final NewTopic newKafkaTopic = new NewTopic(kafkaTopic, 1, (short) 1);
    kafkaAdmin.createTopics(
        Sets.newHashSet(newKafkaTopic),

new CreateTopicsOptions()

sijie

Schema<KeyValue<Integer, String>> kvSchema = Schema.KeyValue(
    Schema.INT32,
    Schema.STRING,
    KeyValueEncodingType.SEPARATED
);

sijie

Consumer<GenericRecord> pulsarConsumer = client.newConsumer(Schema.AUTO_CONSUME())
    …
    .subscribe();
Message<GenericRecord> msg = consumer.receive() ;
GenericRecord record = msg.getValue();

sijie

Producer<GenericRecord> producer = client.newProducer(Schema.generic(schemaInfo)).create();
producer.newMessage()
    .value(schema.newRecordBuilder()
        .set("intField", 32)
        .build())
    .send();

sijie

RecordSchemaBuilder recordSchemaBuilder = SchemaBuilder.record("schemaName");
recordSchemaBuilder
    .field("intField")
    .type(SchemaType.INT32);
SchemaInfo schemaInfo = recordSchemaBuilder.build(SchemaType.AVRO);
Schema<GenericRecord> schema = Schema.generic(schemaInfo);

sijie

Compatibility Check Strategy	Changes allowed	Check against which schemas	Upgrade first
ALWAYS_INCOMPATIBLE	All changes are disabled	All previous versions	None
ALWAYS_COMPATIBLE	All changes are allowed	Compatibility checking disabled	Depends
BACKWARD	Delete fields; Add optional fields	Latest version	Consumers
BACKWARD_TRANSITIVE	Delete fields; Add optional fields	All previous versions	Consumers
FORWARD	Add fields; Delete optional fields	Latest version	Producers
FORWARD_TRANSITIVE	Add fields; Delete optional fields	All previous versions	Producers
FULL	Modify optional fields	Latest version	Any order
FULL_TRANSITIVE	Modify optional fields	All previous versions	Any order

sijie

import (
  "fmt"
  "context"
  "github.com/apache/pulsar/pulsar-function-go/pf"
)

func HandleRequest(ctx context.Context, in []byte) error {
  fmt.Println(string(in) + "!")
 return nil

sijie

Consumer<String> consumer = client.newConsumer(Schema.STRING)
    .topic("my-topic")
    .subscriptionName("my-subscription")
    .replicateSubscriptionState(true)
    .subscribe();

sijie

client.newConsumer()
    .topic("topic")
    .subscriptionType(SubscriptionType.Key_Shared)
    .subscriptionName("key-shared-subscription")
    .subscribe();