pulsar

Can you share how the cluster is setup and how many brokers are there in the cluster. To debug, probably you can get the owner broker of the topic you are using from pulsar-admin and try using that directly.

Did you enable the client side debug log? It shows the proxy and broker address upon redirect.
Also have you tried pulsar-client if you suspect a broker issue?
What is proxy and broker log? Can you see connection from the client

The values that are being added into the map for the integer should have quotes otherwise it is read as integer. Maybe that’s what is going on in this case?

image.png

{"name":"event_data","type":{"type":"map","values":["null","long","string","double","int"]}
although I set this map have multiple types but when I checked it only have map[string]bigInt =&gt; that make my string value in map fail to query

Hello Abhilash, When a Pulsar sink is configured with effectively_once delivery guarantee and it sees an exception during message processing, it will go through a process called "transaction rollback" to ensure that messages are not lost or duplicated. This is the ideal behavior. If you think, this is a bug, can you report this and prolly we can get more inputs into if this is expected behavior..

Thanks <@U03BQF7P61Z>. Transaction rollback is fine but here the Pulsar job kills the sink task which is not ideal. We expect that after failing the record, it will be retried because these errors can be transient. I will raise a issue on Github.

<@U03BQF7P61Z> I raised a github issue <https://github.com/apache/pulsar/issues/19894|here>.

How did you take the backup? Are you backing up the EBS volumes?

Don't judge: this is on kubernetes with the EFS CSI driver on the backend and AWS backup taking snapshots.

Ok you can judge AWS backup as that was my idea, but EFS was not

What you described above is generally correct in theory. However, there are a lot of corner cases that you need to handle when performing a restore operation (i.e. handling zombie zookeeper znodes, handling zombie ledgers, etc).

If you are already on Kubernetes, you can use tools like Velero to automate the process: <https://velero.io/> but you still need to clean things up when performing a restore operation.

Alternatively, there are two approaches to do backup:

1. You can use geo-replication to live "backup" a cluster to a remote location. So you can cut over quickly if one fails.
2. You can use metadata backup + tiered storage as the backup solution. In this approach, you can just configure tiered storage as the data storage and offload the data to S3. So you only keep minimal data in bookkeeper. You then just need to backup zookeeper. In this way, you can quickly spin up a new Pulsar cluster and restore the metadata from backup and you are able to access the data already offloaded to S3. We have automated this as part of a metadata backup/restore pool (see <https://docs.streamnative.io/docs/backup-restore-metadata-tool>)
To choose which backup methods, it is really up to what is your RPO and RTO.

Yeah I was thinking about Velero.  I've implemented it in the past.

It'll take me a sec to read the rest of your message.

I have seen this metadata tool.  I think we should probably be using it but it's not implemented.  I started down implementation but hit snags with volumes somewhere in the helm charts (it's been about a week since I last looked).

It's probably worth attempting again and I'll argue that we should update our copy of the helm charts, in general.  Or stop copying helm charts and use a tag/release of the streamnative charts.

Is there an easy way to tell if things are zombies?

Hello <@U04UWPEN8LV> I agree with Sijie's approach. If you're using k8s, you can use velero to backup

<@U04UWPEN8LV> There is no easy way to tell if things are "zombies". You need to basically audit the metadata and data in order to do that. That was one of the reasons we developed this metadata backup/restore tool to ensure there is a clear state to restore. The drawback of metadata backup is that you need to ensure the data is offloaded to tiered storage. Your RPO is determined by how fast you offload the data to tiered storage.

well we got it to work.  I would not recommend this approach but we did :slightly_smiling_face:

If there are better tools / documentation I'm all ears.

FWIW, the Spring framework has support for Pulsar Consumers.

adding <@ULBJ01G6N> and <@UDZPKSJSK> who are contributing to the Spring for Apache Pulsar Framework

Yes,
• Spring has an integration (cc <@U03RM6GQ2M9>)
    ◦  <https://github.com/spring-projects-experimental/spring-pulsar> . It will soon be GA.
    ◦ It contains starters for Spring Boot
    ◦ It contains a binder for <https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html|Spring Cloud Stream>
• Micronaut has an integration: <https://micronaut-projects.github.io/micronaut-pulsar/latest/guide/>
• Quarkus is building one: <https://github.com/smallrye/smallrye-reactive-messaging/pull/2002>
Also note that there's now a <https://github.com/apache/pulsar-client-reactive|Reactive client> you can use with purely reactive frameworks (eg. Vert.x)

<@U04SMQD7PSP> feel free to ping me if you have any questions on spring-pulsar (the <https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/reference/html/|reference guide> should have most of what you need though).

<@U04SMQD7PSP> pls take a minute to read this article as well

<https://dzone.com/articles/spring-boot-app-connected-to-pulsar-and-cassandra>

Thanks <@U03RM6GQ2M9> for your continued support of spring-pulsar !

You are more than welcome <@U9NHC1ETG> - we always have fun when we do!

{
    "type": "record",
    "namespace": "com.example",
    "name": "test_more_field",
    "fields": [

      { "name": "event_data", "type": {
        "type" : "map",
        "values" : ["null", "int", "long", "double", "string"]}
       }
    ]
}
if my schema like this, and I connect presto with pulsar, does presto support type map to query? because I got this issue when I sent this data to pulsar {
                "data":"abc",
                "field1": 2,
                "field2": 2.3,
            },
```[16842752] Query failed (#20230322_024731_00084_uy4iw): could not parse value 'abc' as 'bigint' for column 'event_data' io.prestosql.spi.PrestoException: could not parse value 'abc' as 'bigint' for column 'event_data'```

hi guys. any help on this issue is greatly appreciated :pray:
<https://github.com/apache/pulsar/issues/19894>

Did you reproduce this issue? Could you share more context to this issue?

Hello Will, can you mention the sample query you use to see if we can try to assist

hi Bala, thank you, here are a couple of my queries...

*presto&gt;* `select * from pulsar."q6/default".bnr_url limit 10;`
```Query 20230324_150739_00004_kcv3t failed: Internal error```
*presto&gt;* `select * from pulsar."q6/default"."bnr-session-function-dlq" limit 10;`
```Query 20230324_150805_00006_kcv3t failed: java.lang.RuntimeException: java.lang.NullPointerException```

When you add a new disk to an existing bookie, you are losing the “cookie” (a unique identifier for the BK cluster) that was persisted onto the old disk. This cookie is used to confirm membership in the BK cluster, and since the new bookie doesn’t have one, it can’t join. It might be best to enable storage expansion and add a new bookie to replace the old one.

Thank you <@U9NHC1ETG> for the suggestion. After changing the bookkeeper configuration to true and restarting the bookie pod, the Readiness probe of new bookie-pod seems to be failing , Is it because there is active data-replication happening in the background ? Or is this indicative of something else being wrong ?
`Readiness probe failed: HTTP probe failed with statuscode: 503` 

can you share the container logs for bookie-0 ?

"2023-03-24T17:50:24,157+0000 [main] ERROR org.apache.bookkeeper.bookie.Bookie - Exception while replaying journals, shutting down\n"

Nothing in specific, but here are some lines from when the pod is trying to restart
```10:53:33.635
[conf/bookkeeper.conf] Applying config journalDirectories = /pulsar/data/bookkeeper/journal
 
10:53:33.635
[conf/bookkeeper.conf] Applying config allowStorageExpansion = true
 
10:53:33.635
[conf/bookkeeper.conf] Applying config autoRecoveryDaemonEnabled = false
 
10:53:33.635
[conf/bookkeeper.conf] Applying config dbStorage_readAheadCacheMaxSizeMb = 512
 
10:53:33.635
[conf/bookkeeper.conf] Applying config dbStorage_rocksDB_blockCacheSize = 16777216
 
10:53:33.635
[conf/bookkeeper.conf] Applying config dbStorage_rocksDB_writeBufferSizeMB = 128
 
10:53:33.635
[conf/bookkeeper.conf] Applying config dbStorage_writeCacheMaxSizeMb = 512
 
10:53:33.635
[conf/bookkeeper.conf] Applying config httpServerEnabled = true
 
10:53:33.635
[conf/bookkeeper.conf] Applying config httpServerPort = 8000
 
10:53:33.635
[conf/bookkeeper.conf] Applying config journalDirectories = /pulsar/data/bookkeeper/journal
 
10:53:33.635
[conf/bookkeeper.conf] Applying config journalMaxBackups = 0
 
10:53:33.635
[conf/bookkeeper.conf] Applying config ledgerDirectories = /pulsar/data/bookkeeper/ledgers
 
10:53:33.635
[conf/bookkeeper.conf] Applying config statsProviderClass = org.apache.bookkeeper.stats.prometheus.PrometheusMetricsProvider
 
10:53:33.635
[conf/bookkeeper.conf] Applying config useHostNameAsBookieID = true
 
10:53:33.635
[conf/bookkeeper.conf] Applying config zkLedgersRootPath = /ledgers
 
10:53:33.635
[conf/bookkeeper.conf] Applying config zkServers = release-prodclone-pulsar-zookeeper:2181
 
10:53:33.635
[conf/bookkeeper.conf] Updating config journalDirectories = /pulsar/data/bookkeeper/journal
 
10:53:33.758
[0.008s][info][gc,heap] Heap region size: 2M
 
10:53:33.758
[0.008s][debug][gc,heap] Minimum heap 2147483648  Initial heap 2147483648  Maximum heap 8589934592
 
10:53:33.770
[0.019s][info ][gc    ] Using G1
 
10:53:33.770
[0.019s][info ][gc,heap,coops] Heap address: 0x0000000600000000, size: 8192 MB, Compressed Oops mode: Zero based, Oop shift amount: 3
 
10:53:33.770
[0.020s][info ][gc,cds      ] Mark closed archive regions in map: [0x00000007ffe00000, 0x00000007ffe6aff8]
 
10:53:33.770
[0.020s][info ][gc,cds      ] Mark open archive regions in map: [0x00000007ffd00000, 0x00000007ffd47ff8]
 
10:53:33.807
[0.057s][info ][safepoint    ] Entering safepoint region: EnableBiasedLocking
 
10:53:33.807
[0.057s][info ][safepoint    ] Leaving safepoint region
 
10:53:33.807
[0.057s][info ][safepoint    ] Total time for which application threads were stopped: 0.0001605 seconds, Stopping threads took: 0.0000475 seconds
 
10:53:33.808
[0.057s][info ][safepoint    ] Application time: 0.0004615 seconds
 
10:53:33.808
[0.057s][info ][safepoint    ] Entering safepoint region: Deoptimize
 
10:53:33.808```

Looks like it is having trouble reading the journal files. Did you replace the PVC for the BOTH the ledger and journal disks? If yes, then it will have trouble coming back online. A better approach might be to just add a fourth bookie and remove bookie-0. Then allow the ledger recovery process populate bookie-3

I actually only replaced the PVC for journal and didn't touch the ledger.
So, I was able to bring up a new bookie, bookie-3 by scaling up the replicaCount to 4 . And deleting bookie-0 manually results in a new pod coming up which is still having trouble reading the journal (the name seems to be an issue ? ) And changing the replicaCount down to 3, deletes bookie-3 instead of bookie-0. Either ways I seem to be stuck with a unhealthy bookie pod :grimacing:

"Then allow the ledger recovery process populate bookie-3"
Also does the ledger recovery process happen automatically or would I need to run `bin/bookkeeper autorecovery`  manually

did you deploy using the Apache helm charts? Do you have a pod named `autorecovery` ?

when you delete bookie-0, start by removing the PVCs associated with it. Otherwise any newly re-spawned `bookie-0` instances will be linked to those volumes due to naming conventions, etc. That’s why you end up with the same issue on the new bookie-0 instance.

<@U9NHC1ETG>, Yes I do have the autorecovery pod in running status. Was just curious to know if I would need to run any manual steps once a new bookie is up and running ?

no, the autorecovery pod will handle the process. manually starting a second process wouldn’t be great.

Hello David, pls take a look at these threads for some good discussion on Key shared subscription

<https://apache-pulsar.slack.com/archives/C5Z4T36F7/p1676397806591399>

<https://apache-pulsar.slack.com/archives/C5Z4T36F7/p1676398580761839>

so is there a better helm chart to use or is this a known issue perhaps

You can try the StreamNative Helm <https://github.com/streamnative/charts|charts> instead. There is also a series of <https://streamnative.io/blog/pulsar-operators-tutorial-part-1-create-apache-pulsar-cluster-kubernetes|blog> posts on installing these charts.

I actually installed pulsar with helm charts last week using this repo from DataStax..didn't seem to have faced any issue

<https://github.com/datastax/pulsar-helm-chart>

The ideal response would be to submit a PR to fix <https://github.com/apache/pulsar-helm-chart>

Looks like we have a fix here: <https://github.com/apache/pulsar-helm-chart/pull/356>.

Hello Arnold, OSS Flink Connector for Pulsar don't support this feature at the moment

There's a PR opened for support so hopefully it will arrive soon: <https://github.com/apache/flink-connector-pulsar/pull/35>

Hello Nick, what version of Pulsar is this

It looks like the subscription name not found, how this exception happened? what operations have you done?

Hi All,
I was checking this PIP <https://github.com/apache/pulsar/wiki/PIP-82:-Tenant-and-namespace-level-rate-limiting>
even before this change there was a way to set namespace publish quota's at a namespace level. (<https://pulsar.apache.org/admin-rest-api/?version=2.7.0#operation/createNamespace>)

This PIP talks about how to enforce rate-limiting using the resourceGroups , so wanted to know how and where was the namespace level publish quota has been used prior to this?
Are these both solving the same use case or this <https://github.com/apache/pulsar/wiki/PIP-82:-Tenant-and-namespace-level-rate-limiting|PIP-82> is adding improvements on the earlier? (one thing I noticed is earlier there was no way to set tenant level quota, but this PIP talks about tenant level quota as well) .

so I assume rsylog files reside in a file system..Although, there is no direct plugin from rsyslog to Pulsar, there is a file source connector.

<https://pulsar.apache.org/docs/next/io-file-source/>

<https://www.youtube.com/watch?v=eu75cmp9Tjk&amp;ab_channel=DataStaxDevelopers>

Please take a look at this and see if it fits your use case

The 5MAP has some examples on how to use the file source

Thanks Bala, I will check this out and see if it fits the use case and scales

This is another possible syslog Pulsar integration <https://github.com/datastax/pulsar-3rdparty-connector/tree/master/pulsar-connectors/camel-syslog>

you will have to modify your helm charts and they config file of your zookeeper cluster should contain all the zookeepers mentioning whether its a observer or a participant. And you need to use the same config file for deploying across all clusters

the content mean :The config store uses local zookeeper cluster??.
i say right ??

Yes, it is suggesting that you can use local zk nodes for global zk as well

Yes I believe so based on <https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L2631|this>

have you run the cluster initialization job ? <https://pulsar.apache.org/docs/2.11.x/admin-api-clusters/#initialize-cluster-metadata>

create pulsar instance  must  "discovery.conf"???

Caused by: org.apache.pulsar.shade.javax.ws.rs.ClientErrorException: HTTP 412 Clusters do not exist

spring boot project,,,i start producer ,show top exception

```@PostConstruct
public void initEnvirment() throws PulsarClientException, PulsarAdminException {
    PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(adminServiceUrl).build();
    if (!admin.tenants().getTenants().contains("my-tenant")) {
        HashSet&lt;String&gt; clusters = new HashSet&lt;&gt;();
        clusters.add("pulsar-instance-1");
        HashSet&lt;String&gt; adminRoles = new HashSet&lt;&gt;();
        adminRoles.add("test-admin-role");
        TenantInfo config = TenantInfo.builder().allowedClusters(clusters).adminRoles(adminRoles).build();
        admin.tenants().createTenant("my-tenant", config);
    }
    if (!admin.namespaces().getNamespaces("my-tenant").contains("my-tenant/my-namespace")) {
        admin.namespaces().createNamespace("my-tenant/my-namespace");
    }
}```


./pulsar initialize-cluster-metadata \
   --cluster pulsar-instance-1 \
   --metadata-store zk:192.168.2.21:2181,192.168.2.45:2181,192.168.2.48:2181,192.168.2.101:2181,192.168.2.102:2181,192.168.2.103:2181/pulsar-instance-1 \
   --configuration-metadata-store zk:192.168.2.21:2184,192.168.2.45:2184,192.168.2.48:2184,192.168.2.101:2184,192.168.2.102:2184,192.168.2.103:2184/pulsar-instance-1 \
   --web-service-url <http://192.168.2.21:8080>,192.168.2.45:8080,192.168.2.48:8080,192.168.2.101:8080,192.168.2.102:8080,192.168.2.103:8080/ \
   --web-service-url-tls <https://192.168.2.21:8443>,192.168.2.45:8443,192.168.2.48:8443,192.168.2.101:8443,192.168.2.102:8443,192.168.2.103:8443/ \
   --broker-service-url <pulsar://192.168.2.21:6650>,192.168.2.45:6650,192.168.2.48:6650,192.168.2.101:6650,192.168.2.102:6650,192.168.2.103:6650/ \
   --broker-service-url-tls <pulsar+ssl://192.168.2.21:6651>,192.168.2.45:6651,192.168.2.48:6651,192.168.2.101:6651,192.168.2.102:6651,192.168.2.103:6651/

the ./pulsar initialize-cluster-metadata must be run once.
what is the output? maybe it failed. can you run it again? it’s idempotent

first step run :./bookkeeper shell metaformat   ???

second step run :   initialize-cluster-metadata ？？

no, first initialize-cluster-metadata, then bookkeeper shell metaformat for each bookie

again run initialize-cluster-metadata,,, run 2 time ,,,how do ??

run init command log show :
 INFO  org.apache.pulsar.PulsarClusterMetadataSetup - Cluster metadata for 'pulsar-instance-1' setup correctly

H_P)}[7]LY%4MAUSLUK)7@0.png

i start bookkeeper in pulsar instance,,, show bottom error:

Caused by: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /ledgers/LAYOUT

You need to initiate the Pulsar cluster’s metadata first.
<https://pulsar.apache.org/docs/2.11.x/deploy-bare-metal-multi-cluster/#cluster-metadata-initialization>

Or maybe it’s even happening when an old ledger is closed

the lastActive is not persisted, so if it the topic is frequently unloaded it will not expire

how is lastActive determined? Would moving the cursor due to a ledger being closed also stop the subscription from expiring?

It would be the timestamp when the last consumer has disconnected from the subscription

But it’s not persisted? So then what does that timestamp get set to when the topic changes ownership? I didn’t see any reference to lastActive in the topic’s `stats` or `stat-internal`

If we load those topics to any brokers, the subscription expiration will be triggered.
<https://github.com/apache/pulsar/issues/18128>
We have one proposal to load those topics and the feature will be released in Pulsar 3.0

Check out <https://github.com/datastax/pulsar-transformations|https://github.com/datastax/pulsar-transformations>, which is a built-in transform function 

<@U0514CX629Z> To get a list of the available source connectors, you can run the command `./bin/pulsar-admin sources available-sources` and for sinks `./bin/pulsar-admin sinks available-sinks` for more details on how to configure these built-in connectors you can go to the <https://pulsar.apache.org/docs/2.11.x/io-connectors/|documentation>

For functions, there aren’t any built-in ones in OSS pulsar. You have to place the function’s artifact in the `functions` directory of the function workers before startup in order to reference them in this manner.

<@U9NHC1ETG> thanks for that, I’m aware of the connectors, just puzzled by the messaging of the built-in functions :slightly_smiling_face:

Thanks for the details on the functions though

I’ve created a couple already but in a chicken and egg situation with the Debezium PostgreSQL Source which creates it’s own topics that I want to apply the functions to.

No worries. I saw that you said “function” but then you mentioned Debezium, so I thought maybe you were asking about connectors.  :smiley:

Yeah it’s a combination. The transforms don’t work for Debezium in Pulsar (only Kafka) ergo I want to add a function but you can only apply the function after deploying the connector that needs the function to route the messages…. :infinity: lol

So these are <https://debezium.io/documentation/reference/stable/transformations/index.html|SMTs> ?

I believe SMTs don’t work in Pulsar, well the way I’m interpreting the connectors abilities.

The repo I mentioned at the beginning of this thread is a Pulsar function designed to give similar capabilities to Kafka SMT. 

Yeah I saw that <@UDPUMMMS5>, going through it now.

Just trying to decode the Debezium Schema to the AVRO schema with this SMT-ish Function. See if I can make it work with the “cast” feature.

Functions like the transform one can be run inside a sink connector, but not a source one. That’s how we often use the one above.

Correct, so you would setup the Debezium source to publish to a topic and then configure a Pulsar Function to ingest those messages and perform the SMT -like processing. But you will have to deploy the functions with the `./bin/pulsar-admin functions create` command and specify the --jar or --py

Debezium creates it’s own topics so it will depend on a couple of the environmental factors (db name, partition, tablename)

We package the function inside a Docker image so you don’t have to specify the —jar. There is a specific directory Pulsar will look for it in.

So yeah I think I’ll have to just make sure I can deploy the function to all the table topics it creates that I need to re-route.

Maybe a regex pattern for the input topic config. Or as Chris mentioned, if this is a Debezium Sink you can run it inside the connector

Cool, I haven’t tried docker containers yet. Part of the next step.

Np and good luck. Thanks for pointing out this library <@UDPUMMMS5>, really cool collection of functions.

(cc <@UDZPKSJSK> <@U03LWBY199T> who are the maintainers of <https://github.com/datastax/pulsar-transformations> )

&gt; Functions like the transform one can be run inside a sink connector, but not a source one. That’s how we often use the one above.
Just want to mention that the possibility of running the function in-memory in the sink is not available in Pulsar 2.11. It will be part of the next version.
We (DataStax) use our own fork (Luna Streaming) that has it already available. It's the exact same code as what will be available in the next Pulsar version.
I'm saying this just to avoid confusion.

btw. Is there any discussion about how "real time" or "near real time" is defined? What type of end-to-end latencies for example? What's your insight on that?

the definitions are all over the place seems like a few hundred ms is considered real time

how about "near real time"? :slightly_smiling_face:

well that’s spark micro batch latency now, probably around a few seconds.

I've heard even a 5 minute processing latency to be referred to as "near real time". I guess this is where there's lots of different opinions of what it is. One possible reason might be that in some contexts it might not matter if the latency is 1 second or 5 minutes, as long as it's "near real time"? It would be great to have more examples and comparisons where it matters and where not.

In my experience, the term “real-time” was reserved for applications such as a self-driving car, where the processing has to occur fast enough (if not faster than a human) to make a decision and react, e.g. hit the brakes, turn the wheel, etc. in order to avoid a collision Consistency is paramount, and any high latency events were intolerable.  But, as you both pointed out, that definition has been stretched quite a bit to fit marketing campaigns…..

```what is the difference between real-time and near real-time?
Real-time and near real-time refer to different levels of timeliness in data processing and communication.

Real-time processing refers to the ability of a system to process and deliver data immediately as it is generated, without any delay. This means that the system can respond to events and actions as they occur in real-time, and users can receive information and updates instantly. For example, a real-time stock market data feed provides up-to-the-second information on stock prices and market trends.

Near real-time processing, on the other hand, refers to the ability of a system to process and deliver data with a minimal delay, typically within a few seconds or minutes of its generation. This means that users can still receive timely updates, but with a slight delay compared to real-time processing. For example, a near real-time weather forecasting system might provide updated weather forecasts every 10 or 15 minutes.

In summary, the main difference between real-time and near real-time processing is the amount of delay or latency involved in processing and delivering data. Real-time processing is immediate, while near real-time processing involves a slight delay but still provides timely information.```

my understanding of real time doesn’t involve “internet” speed but electronic/light speed.

computer time sync is still not flawless to satisfy “realtime”

Registration for the Pulsar Virtual Summit Europe is open! <https://events.zoom.us/ev/Ap6rsDg9LeVfmdajJ_eB13HH026J1d_o8OoTKkQnl_jzVl-srhwB~AggLXsr32QYFjq8BlYLZ5I06Dg|Register for free here>
Slack Conversation

Journal is write ahead log. it is not replicated. the replicated data are ledgers. if you check the data write path, broker will send ack back to producer when data written to journal (but not in ledger yet).  if you use local ssd, there is a chance that the data in broker memory are not persist to ledger disk and broker will need to replay journal to mend the gap.  if gke replica this bookie to other nodes, there is no journal to replay.     

for gke, you can add more journal disks using premium pd to increase the throughput. 

autorecovery is checking metadata in zk and find any underreplicate ledgers. it doesn’t replay journal. 

<@U0514S0P9C6> There are 2 options for the journal:

1. disable fsync — `journalSyncData=false`
2. disable journal all-together — `journalWriteData=false`
The 2nd option makes more sense instead of putting on an ephemeral volume

This is really helpful. Thanks for your help!

I have the base config in my yaml:

```inputSpecs:
  "<persistent://my-tenant/my-namespace/dbserver1.public.person>":
    schemaType: "key_value"```
I think it’s the schemaProperties but there is so little documentation on how to set it.

My function logs state the schema is:
```schemaInfo: {
  "name": "KeyValue",
  "schema": {
    "key": {
      "name": "Bytes",
      "schema": "",
      "type": "BYTES",
      "timestamp": 0,
      "properties": {}
    },
    "value": {
      "name": "Bytes",
      "schema": "",
      "type": "BYTES",
      "timestamp": 0,
      "properties": {}
    }
  },
  "type": "KEY_VALUE",
  "timestamp": 0,
  "properties": {
    "key.schema.name": "Bytes",
    "key.schema.properties": "{}",
    "key.schema.type": "BYTES",
    "kv.encoding.type": "INLINE",
    "value.schema.name": "Bytes",
    "value.schema.properties": "{}",
    "value.schema.type": "BYTES"
  }
}```

Querying the topic I get the Debezium schema (real long fyi)

```$ bin/pulsar-admin schemas get <persistent://my-tenant/my-namespace/dbserver1.public.person>
{
  "version": 0,
  "schemaInfo": {
    "name": "dbserver1.public.person",
    "schema": {
      "key": {
        "name": "KafKaJson",
        "schema": {
          "type": "record",
          "name": "Key",
          "namespace": "dbserver1.public.person",
          "fields": [
            {
              "name": "id",
              "type": "int"
            }
          ],
          "connect.name": "dbserver1.public.person.Key"
        },
        "type": "JSON",
        "timestamp": 0,
        "properties": {
          "__AVRO_READ_OFFSET__": "0"
        }
      },
      "value": {
        "name": "KafKaJson",
        "schema": {
          "type": "record",
          "name": "Envelope",
          "namespace": "dbserver1.public.person",
          "fields": [
            {
              "name": "before",
              "type": [
                "null",
                {
                  "type": "record",
                  "name": "Value",
                  "fields": [
                    {
                      "name": "id",
                      "type": "int"
                    },
                    {
                      "name": "first_name",
                      "type": "string"
                    },
                    {
                      "name": "last_name",
                      "type": [
                        "null",
                        "string"
                      ]
                    },
                    {
                      "name": "email",
                      "type": [
                        "null",
                        "string"
                      ]
                    }
                  ],
                  "connect.name": "dbserver1.public.person.Value"
                }
              ]
            },
            {
              "name": "after",
              "type": [
                "null",
                "Value"
              ]
            },
            {
              "name": "source",
              "type": {
                "type": "record",
                "name": "Source",
                "namespace": "io.debezium.connector.postgresql",
                "fields": [
                  {
                    "name": "version",
                    "type": "string"
                  },
                  {
                    "name": "connector",
                    "type": "string"
                  },
                  {
                    "name": "name",
                    "type": "string"
                  },
                  {
                    "name": "ts_ms",
                    "type": "long"
                  },
                  {
                    "name": "snapshot",
                    "type": [
                      {
                        "type": "string",
                        "connect.version": 1,
                        "connect.parameters": {
                          "allowed": "true,last,false"
                        },
                        "connect.default": "false",
                        "connect.name": "io.debezium.data.Enum"
                      },
                      "null"
                    ],
                    "default": "false"
                  },
                  {
                    "name": "db",
                    "type": "string"
                  },
                  {
                    "name": "sequence",
                    "type": [
                      "null",
                      "string"
                    ]
                  },
                  {
                    "name": "schema",
                    "type": "string"
                  },
                  {
                    "name": "table",
                    "type": "string"
                  },
                  {
                    "name": "txId",
                    "type": [
                      "null",
                      "long"
                    ]
                  },
                  {
                    "name": "lsn",
                    "type": [
                      "null",
                      "long"
                    ]
                  },
                  {
                    "name": "xmin",
                    "type": [
                      "null",
                      "long"
                    ]
                  }
                ],
                "connect.name": "io.debezium.connector.postgresql.Source"
              }
            },
            {
              "name": "op",
              "type": "string"
            },
            {
              "name": "ts_ms",
              "type": [
                "null",
                "long"
              ]
            },
            {
              "name": "transaction",
              "type": [
                "null",
                {
                  "type": "record",
                  "name": "ConnectDefault",
                  "namespace": "org.apache.pulsar.kafka.shade.io.confluent.connect.avro",
                  "fields": [
                    {
                      "name": "id",
                      "type": "string"
                    },
                    {
                      "name": "total_order",
                      "type": "long"
                    },
                    {
                      "name": "data_collection_order",
                      "type": "long"
                    }
                  ]
                }
              ]
            }
          ],
          "connect.name": "dbserver1.public.person.Envelope"
        },
        "type": "JSON",
        "timestamp": 0,
        "properties": {
          "__AVRO_READ_OFFSET__": "0"
        }
      }
    },
    "type": "KEY_VALUE",
    "timestamp": 1680128742019,
    "properties": {
      "key.schema.name": "KafKaJson",
      "key.schema.properties": "{\"__AVRO_READ_OFFSET__\":\"0\"}",
      "key.schema.type": "JSON",
      "kv.encoding.type": "SEPARATED",
      "value.schema.name": "KafKaJson",
      "value.schema.properties": "{\"__AVRO_READ_OFFSET__\":\"0\"}",
      "value.schema.type": "JSON"
    }
  }
}```

Okay I’ve made some progress here, it turns out that Debezium uses mixed schema types for key_value schemas. I’ve started my journey to dissecting schemas and trying to create my own function and making some progress.

The pulsar-transformations function is a good learning tool but it seems there are some basic limitations that aren’t working for me.