https://pulsar.apache.org/ logo
Join Slack
Powered by
This message was deleted.
# general
s
This message was deleted.
y
do you have stats-internal?
o
sorry, what does that mean?
or what is it? haha
These are the stats written to the log if that helps:
Copy code
2023-05-17T20:19:35,142+0000 [pulsar-timer-6-1] INFO  org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl - [<persistent://hyperflow/demo/demo2_sink_topic>] [hyperflow/demo/demo2_sink] [05dee] Prefetched messages: 0 --- Consume throughput received: 0.02 msgs/s --- 0.00 Mbit/s --- Ack sent rate: 0.00 ack/s --- Failed messages: 0 --- batch messages: 0 ---Failed acks: 0
y
pulsar-admin topics stats-internal persistent://test-tenant/ns1/tp1
o
Oh awesome! Thats new to me so thanks for showing it to me. Here is the output:
Copy code
{
  "entriesAddedCounter": 1,
  "numberOfEntries": 1,
  "totalSize": 90,
  "currentLedgerEntries": 1,
  "currentLedgerSize": 90,
  "lastLedgerCreatedTimestamp": "2023-05-17T20:16:35.195Z",
  "waitingCursorsCount": 1,
  "pendingAddEntriesCount": 0,
  "lastConfirmedEntry": "311:0",
  "state": "LedgerOpened",
  "ledgers": [
    {
      "ledgerId": 311,
      "entries": 0,
      "size": 0,
      "offloaded": false,
      "underReplicated": false
    }
  ],
  "cursors": {
    "hyperflow%2Fdemo%2Fdemo2_sink": {
      "markDeletePosition": "311:-1",
      "readPosition": "311:1",
      "waitingReadOp": true,
      "pendingReadOps": 0,
      "messagesConsumedCounter": 0,
      "cursorLedger": 312,
      "cursorLedgerLastEntry": 0,
      "individuallyDeletedMessages": "[]",
      "lastLedgerSwitchTimestamp": "2023-05-17T20:16:35.224Z",
      "state": "Open",
      "numberOfEntriesSinceFirstNotAckedMessage": 2,
      "totalNonContiguousDeletedMessagesRange": 0,
      "subscriptionHavePendingRead": true,
      "subscriptionHavePendingReplayRead": false,
      "properties": {}
    }
  },
  "schemaLedgers": [
    {
      "ledgerId": 307,
      "entries": 1,
      "size": 80,
      "offloaded": false,
      "underReplicated": false
    },
    {
      "ledgerId": 308,
      "entries": 1,
      "size": 342,
      "offloaded": false,
      "underReplicated": false
    },
    {
      "ledgerId": 313,
      "entries": 1,
      "size": 1121,
      "offloaded": false,
      "underReplicated": false
    }
  ],
  "compactedLedger": {
    "ledgerId": -1,
    "entries": -1,
    "size": -1,
    "offloaded": false,
    "underReplicated": false
  }
}
I managed to get it working but it required rebuilding the postgres jdbc sink with 
_properties_.*setProperty*("stringtype", "unspecified");
so it would convert strings to `text`rather than 
character varying
(idea stolen from this thread: https://postgrespro.com/list/thread-id/1939980) and then sending the datetime as a string rather than a datatime object. it works but I would rather use Pulsar built packages rather than my own so if anyone knows a way to do it please let me know!!
Open in Slack
PreviousNext
Powered by