Provide documentation for correct usage of HTTP sink

[jepetko]

For integration of our existing REST APIs with Apache Pulsar (version 3.2.2) we want to utilize HTTP sink.

First, we configured the sink using processingGuarantee: atmost_once and autoAck: true:

apiVersion: compute.functionmesh.io/v1alpha1
kind: Sink
metadata:
  name: my-sink
spec:
  tenant: my-tenant
  namespace:  my-namespace
  className: org.apache.pulsar.io.http.HttpSink
  image: 'docker..../...'
  java:
    jar: /pulsar/connectors/pulsar-io-http-3.1.0.6.nar
  input:
    topics:
      - "some-tenant/some-namespace/some-topic"

    typeClassName: org.apache.pulsar.client.impl.schema.GenericObjectWrapper

  subscriptionPosition: latest
  processingGuarantee: atmost_once
  autoAck: true

However, if the URL called by HTTP sink fails, the message should not be acknowledged, hence the message should be redelivered.

For that purpose, we changed the Sink configuration as follows:

  processingGuarantee: effectively_once   # was before atmost_once
  autoAck: false  # was before true

We started to receive the same message multiple times. Also the topic's internal-stats command shows that

• there are 656 not-acknowledged messages (numberOfEntriesSinceFirstNotAckedMessage: 656)
• there are lot of ledgers (this might be related to the big amount of not-acknowledged messages "partitioned" across multiple ledgers)
• message counter is negative (messagesConsumedCounter: -202). Question: is the negative value related to so called "negative acknowledgement?, see

  pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

  Line 807 in 137df29

  public void negativeAcknowledge(MessageId messageId) {

• cursor's status is "NoLedger"

# pulsar-admin  persistent stats-internal my-tenant/my-namespace/my-topic
{
  "entriesAddedCounter" : 453,
  "numberOfEntries" : 655,
  "totalSize" : 242499,
  "currentLedgerEntries" : 19,
  "currentLedgerSize" : 9419,
  "lastLedgerCreatedTimestamp" : "2024-11-04T09:12:52.309Z",
  "waitingCursorsCount" : 1,
  "pendingAddEntriesCount" : 0,
  "lastConfirmedEntry" : "775409:18",
  "state" : "LedgerOpened",
  "ledgers" : [ {
    "ledgerId" : 734396,
    "entries" : 1,
    "size" : 288,
    "offloaded" : false,
    "underReplicated" : false
  }, {
    "ledgerId" : 738628,
    "entries" : 115,
    "size" : 33120,
    "offloaded" : false,
    "underReplicated" : false
  }, (... 35 more ledgers ...) ],
  "cursors" : {
    "my-tenant%2Fmy-namespace%2Fmy-sink" : {
      "markDeletePosition" : "619941:44",
      "readPosition" : "775409:19",
      "waitingReadOp" : true,
      "pendingReadOps" : 0,
      "messagesConsumedCounter" : -202,
      "cursorLedger" : -1,
      "cursorLedgerLastEntry" : -1,
      "individuallyDeletedMessages" : "[]",
      "lastLedgerSwitchTimestamp" : "2024-10-27T02:46:44.66Z",
      "state" : "NoLedger",
      "active" : true,
      "numberOfEntriesSinceFirstNotAckedMessage" : 656,
      "totalNonContiguousDeletedMessagesRange" : 0,
      "subscriptionHavePendingRead" : true,
      "subscriptionHavePendingReplayRead" : false,
      "properties" : { }
    }
  },
  "schemaLedgers" : [ {
    "ledgerId" : 140,
    "entries" : 1,
    "size" : 4181,
    "offloaded" : false,
    "underReplicated" : false
  }, {
    "ledgerId" : 1486,
    "entries" : 1,
    "size" : 2137,
    "offloaded" : false,
    "underReplicated" : false
  } ],
  "compactedLedger" : {
    "ledgerId" : -1,
    "entries" : -1,
    "size" : -1,
    "offloaded" : false,
    "underReplicated" : false
  }
}

When we analyzed the Pulsar source code, we stumbled upon this issue #15560. Although the issue does not deal with sinks in particular, it refers to the combination of processingGuarantee and autoAck parameters. Indeed, autoAck parameter is deprecated as the way how/when the message is acknowledged depends on the processingGuarantee parameter.

Our motivation for autoAck: false and processingGuarantee: effectively_once was that the message gets only acknowledged if the sink succeeds processing (the called REST endpoint returns 2xx). However, it looks like the messages are not acknowledged at all.

When we look at the implementation of the HTTP sink (https://github.com/apache/pulsar/blob/master/pulsar-io/http/src/main/java/org/apache/pulsar/io/http/HttpSink.java#L92), we cannot see any manual acknowledgment. On the other hand, the Cassandra sink has manual acknowledgment in place, see https://github.com/apache/pulsar/blob/master/pulsar-io/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java#L78.

Our conclusion is that autoAck: false does not work in combination with HTTP sink.

• Could you please confirm that?
• Could you also provide more details about valid configurations for the related sink in the official documentation?
• Could you also guide us regarding the handling of failed http sink calls (basically, how do we retry only those messages which fail to be dispatched via http sink to the configured url)?