This message was deleted.

Since this is a Pulsar IO connector, you will have to submit it using the pulsar-admin API for connectors documented here, and in greater detail here.

@David K I have tried but I get this error when I check status

{
  "numInstances": 1,
  "numRunning": 0,
  "instances": [
    {
      "instanceId": 0,
      "status": {
        "running": false,
        "error": "UNAVAILABLE: io exception",
        "numRestarts": 0,
        "numReadFromPulsar": 0,
        "numSystemExceptions": 0,
        "latestSystemExceptions": [],
        "numSinkExceptions": 0,
        "latestSinkExceptions": [],
        "numWrittenToSink": 0,
        "lastReceivedTime": 0,
        "workerId": "c-standalone-fw-localhost-8080"
      }
    }
  ]
}

Can you check the function worker logs for errors? The “UNAVAILABLE: io exception” in the error field looks suspicious. Perhaps the nar file is available on the function-worker node?

@David K please do you know the path to the log? because I have been looking for it

Is the function-worker running as a K8s pod or a standalone process on a physical server? If the former, then

kubectl logs <POD> --tail

should work. If the latter, I am not sure of the topic of my head.

running on standalone

Inside docker or from the command line?

command line

Then the log messages should be in wherever you redirect stdout and stderr to.

ok, it should be in logs/functions

wireshark to look at the network traffic would be your best bet

do you think by not including the authentication header in my config file might have caused this ?

So everything is working now? 🤞

yes boss, just one last clarification

There is an REST client available. Would that work?

yeah I know of this, the app I am talking is a RocketChat(RC) app, so it seems RC does not accept neither ws nor pulsar protocol just their inbuilt http even axios and fetch does not work

Maybe Pulsar Beam would be a better option?

thank you so very much boss, I will check it out

You can use whatever library you like to post to the HTTP endpoint exposed by BEAM

Thank you so much sir

The pulsar protocol does’t have built-in support for this behavior. However, there are two ways to emulate this behavior. The first is to use Pulsar message properties to tag the message with its intended recipient’s ID. Consumers would then consume all the messages, but only process those that are intended for it. The downsides to this approach are that it is inefficient (all messages get sent over the network when only a fraction of them are processed), and that the data is exposed to all consumers, so sensitive data cannot be sent this way. You can solve the data security issue by using per-message encryption to control how can decrypt the message payload.

the very first approach sounds great, what flag can I use to add this properties ?

I would recommend using the message properties, which is just a key/value map. Then pick a hard-coded key value to use for passing this information, e.g. “MSG_RECIPIENT” , etc.

oh thanks, the entire idea I am trying to implement is to control a certain customer's message production rate using

--dispatch-rate-period, -dt

and

--msg-dispatch-rate, -md

in a topic called

public/default/messenger

which has so many other subscribers, so I am just trying to figure out how I can do this with pulsar

The only back pressure mechanism that Pulsar has to control/throttle producers are backlog quotas, but those are at the topic level and not the producer level.

ok, one more question please

It should allow you to use whatever tenant/namespace you like. W/o seeing the error, it sounds like a permissions issue. Did you grant yourself produce/consume and function permissions ?

no I didn’t, when is this permission set ?

It is set at the namespace or topic level. What is the error you are seeing when you try to submit the sink

please what is it that I am doing wrong here

const axios = require('axios');
const FormData = require('form-data');
let data = new FormData();
data.append('classname', 'org.example.MySinkTest');
data.append('sourceSubscriptionName', 'Latest');
data.append('inputs', '["<persistent://public/default/ojoxdan>"]');
data.append('topicsPattern', 'public/default/');
data.append('inputSpecs', '{"schemaType": "type-x", "serdeClassName": "name-x", "isRegexPattern": true, "receiverQueueSize": 5}');
data.append('configs', '{"url":"<http://localhost:9000/consume/sms/smsenvio>"}');
data.append('secrets', 'hello');
data.append('parallelism', '1');
data.append('processingGuarantees', 'EFFECTIVELY_ONCE');
data.append('retainOrdering', 'true');
data.append('resources', '{"cpu":1,"ram":2,"disk":3}');
data.append('autoAck', 'true');
data.append('timeoutMs', '1000');
data.append('cleanupSubscription', 'true');
data.append('runtimeFlags', 'hello');
data.append('archive', '<https://dlcdn.apache.org/pulsar/pulsar-3.0.0/connectors/pulsar-io-http-3.0.0.nar>');

let config = {
  method: 'post',
  maxBodyLength: Infinity,
  url: '<http://localhost:8080/admin/v3/sinks/public/default/msgs>',
  headers: { 
    'Content-Type': 'multipart/form-data', 
    ...data.getHeaders()
  },
  data : data
};

axios.request(config)
.then((response) => {
  console.log(JSON.stringify(response.data));
})
.catch((error) => {
  console.log(error);
});

no matter what I add to this, it always return this response

{
  "reason": "Sink config is not provided"
}

@ojoxdan I’m not super familiar with js but I think it happens because your sinkConfig is not really passed as a

multipart/form-data

form param in the request

@Alexander Preuß I used postman but got the same result. I had the content type set to application/json and used formdata for the data

@ojoxdan this is how the from parameters need to be passed:

curl --location '<http://localhost:8080/admin/v3/sinks/public/default/my-jdbc>' \
--header 'Content-Type: multipart/form-data' \
--form 'sinkConfig="{"tenant":"public", "namespace":"default", ... }"

@Alexander Preuß let me try this out asap because I need it like right now

The backend expects the sinkConfig to be a single form parameter. I believe what happens in your code is it creates a form parameter for every property individually (the documentation/swagger definition is also wrong on the pulsar site regarding this)

ok I am trying now

I think you are still missing a

}

which could cause a parsing error

here again

@ojoxdan I believe you should be able to fix this one by removing the

classname

config propertiy

ok let me try

Are there any errors in the broker/function worker logs around the time of that you post this request?

no sure

It looks like a packaging issue. If you examine the nar file contents, do you see the HttpSinkConfig class?

from here https://pulsar.apache.org/download/

Sorry, any luck with what? Rebuilding the NAR to include the missing class? Fixing the slowness of the httpSink in general?

I downloaded the file from the official repo