https://pulsar.apache.org/ logo
Join Slack
Powered by
Dear Pulsar Community, I am currently working on s...
# general
s
Dear Pulsar Community, I am currently working on setting up Apache Pulsar as an Event Broker, which requires long term storage of messages. Currently I try to manually offload a topic via the Admin API. The API command triggers successfully the offloading and afterwards the following status can be retrieved:
Copy code
{
  "status": "SUCCESS",
  "lastError": "",
  "firstUnoffloadedMessage": {
    "ledgerId": 66,
    "entryId": 0,
    "partitionIndex": -1,
    "batchIndex": -1,
    "batchSize": 0
  }
}
While on the pulsar broker in charge of the topic the following log entries can be found:
Copy code
[pulsar-web-38-3] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [<persistent://offloadTenant/offloading/test>] Starting offload operation at messageId 66:1:-1
[pulsar-web-38-3] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [offloadTenant/offloading/persistent/test] Start ledgersOffload. ledgers=[66, 69] totalSize=87744
[pulsar-web-38-3] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [offloadTenant/offloading/persistent/test] No ledgers to offload
[pulsar-web-38-3] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [<persistent://offloadTenant/offloading/test>] Completed successfully offload operation at messageId 66:1:-1
I have checked the code on Github, which indicates that the offloading is aborted due to a lack of ledger entries, while the ledger keys are known. There are entries within the ledger and I am able to read them through a Python client. On this grounds I suspect, that I may have some misconfiguration going on. This is the current configuration for the namespace:
Copy code
{
  "auth_policies": {
    "namespace_auth": {},
    "destination_auth": {},
    "subscription_auth_roles": {}
  },
  "replication_clusters": [
    "pulsar"
  ],
  "bundles": {
    "boundaries": [
      "0x00000000",
      "0x40000000",
      "0x80000000",
      "0xc0000000",
      "0xffffffff"
    ],
    "numBundles": 4
  },
  "backlog_quota_map": {},
  "clusterDispatchRate": {},
  "topicDispatchRate": {},
  "subscriptionDispatchRate": {},
  "replicatorDispatchRate": {},
  "clusterSubscribeRate": {},
  "publishMaxMessageRate": {},
  "latency_stats_sample_rate": {},
  "retention_policies": {
    "retentionTimeInMinutes": -1,
    "retentionSizeInMB": -1
  },
  "deleted": false,
  "encryption_required": false,
  "subscription_auth_mode": "None",
  "offload_threshold": 0,
  "offload_threshold_in_seconds": -1,
  "schema_compatibility_strategy": "UNDEFINED",
  "is_allow_auto_update_schema": true,
  "schema_validation_enforced": false,
  "offload_policies": {
    "offloadersDirectory": "./offloaders",
    "managedLedgerOffloadDriver": "S3",
    "managedLedgerOffloadMaxThreads": 2,
    "managedLedgerOffloadPrefetchRounds": 1,
    "managedLedgerOffloadThresholdInBytes": 0,
    "managedLedgerOffloadedReadPriority": "TIERED_STORAGE_FIRST",
    "managedLedgerExtraConfigurations": {},
    "s3ManagedLedgerOffloadMaxBlockSizeInBytes": 67108864,
    "s3ManagedLedgerOffloadReadBufferSizeInBytes": 1048576,
    "s3ManagedLedgerOffloadRoleSessionName": "pulsar-s3-offload",
    "gcsManagedLedgerOffloadMaxBlockSizeInBytes": 67108864,
    "gcsManagedLedgerOffloadReadBufferSizeInBytes": 1048576,
    "managedLedgerOffloadBucket": "apache-pulsar-offload",
    "managedLedgerOffloadRegion": "*region-code*",
    "managedLedgerOffloadServiceEndpoint": "*address*",
    "s3Driver": true,
    "gcsDriver": false,
    "fileSystemDriver": false
  },
  "subscription_types_enabled": [],
  "properties": {}
}
Since I am new to Apache Pulsar, I am bit overwhelmed. I hope you can give me some hints on where the problem could lie or how to better debug this problem. If you need more information feel free to ask :) Best wishes, Simon
d
Can you share the output of 
topic stats-internal
for the topic?
s
Copy code
{
  "entriesAddedCounter": 0,
  "numberOfEntries": 2000,
  "totalSize": 87744,
  "currentLedgerEntries": 0,
  "currentLedgerSize": 0,
  "lastLedgerCreatedTimestamp": "2024-05-31T08:21:27.079Z",
  "waitingCursorsCount": 0,
  "pendingAddEntriesCount": 0,
  "lastConfirmedEntry": "66:1999",
  "state": "LedgerOpened",
  "ledgers": [
    {
      "ledgerId": 66,
      "entries": 2000,
      "size": 87744,
      "offloaded": false,
      "underReplicated": false
    },
    {
      "ledgerId": 69,
      "entries": 0,
      "size": 0,
      "offloaded": false,
      "underReplicated": false
    }
  ],
  "cursors": {
    "sub": {
      "markDeletePosition": "66:999",
      "readPosition": "66:1000",
      "waitingReadOp": false,
      "pendingReadOps": 0,
      "messagesConsumedCounter": -1000,
      "cursorLedger": -1,
      "cursorLedgerLastEntry": -1,
      "individuallyDeletedMessages": "[]",
      "lastLedgerSwitchTimestamp": "2024-05-31T08:21:27.119Z",
      "state": "NoLedger",
      "active": false,
      "numberOfEntriesSinceFirstNotAckedMessage": 1,
      "totalNonContiguousDeletedMessagesRange": 0,
      "subscriptionHavePendingRead": false,
      "subscriptionHavePendingReplayRead": false,
      "properties": {}
    },
    "__compaction": {
      "markDeletePosition": "66:1999",
      "readPosition": "66:2000",
      "waitingReadOp": false,
      "pendingReadOps": 0,
      "messagesConsumedCounter": 0,
      "cursorLedger": 71,
      "cursorLedgerLastEntry": 2,
      "individuallyDeletedMessages": "[]",
      "lastLedgerSwitchTimestamp": "2024-05-31T12:56:36.177Z",
      "state": "Open",
      "active": false,
      "numberOfEntriesSinceFirstNotAckedMessage": 1,
      "totalNonContiguousDeletedMessagesRange": 0,
      "subscriptionHavePendingRead": false,
      "subscriptionHavePendingReplayRead": false,
      "properties": {
        "CompactedTopicLedger": 70
      }
    }
  },
  "schemaLedgers": [],
  "compactedLedger": {
    "ledgerId": 70,
    "entries": 2000,
    "size": 135616,
    "offloaded": false,
    "underReplicated": false
  }
}
d
The internal stats indicate that ledger 66 wasn't offloaded. So the success message you received from the administration API call was incorrect
You also have an open subscription on ledger 66, which might be causing the offload issue. Try offloading it again. If that doesn't work, try deleting the subscription named sub, and retry
s
I have tried to offload it again and received the success message, without any offloading happening. Then I deleted the subscription and offloaded the ledger again to no success. The logs of the broker have the following entries:
Copy code
[pulsar-web-38-5] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [offloadTenant/offloading/persistent/test] Start ledgersOffload. ledgers=[66, 69, 72] totalSize=1186686
[pulsar-web-38-5] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [offloadTenant/offloading/persistent/test] No ledgers to offload
[pulsar-web-38-5] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [<persistent://offloadTenant/offloading/test>] Completed successfully offload operation at messageId 66:1:-1
d
What was the command you ran to trigger the offload?
Could you try using the—w switch to force the command to wait for the offload to be completed before returning success?
s
I used curl -vX PUT http://[ip-address]/admin/v2/persistent/offloadTenant/offloading/test/offload -H 'Content-Type: application/json' -d '{"ledgerId":66"entryId":1,"partitionIndex":-1,"batchIndex":-1,"batchSize":0}'
👍 1
d
Can you try adding the 
-w
switch to the command and share the results?
s
I am working on it
Copy code
Error in offload
null

Reason: Error offloading: org.apache.bookkeeper.mledger.ManagedLedgerException$OffloadInProgressException: Offload operation already running
🤔 1
d
We might have to wait a bit…..
👍 1
s
Copy code
Offload was a success
No data was uploaded to the S3 Storage and server logs are the same as before
Copy code
2024-06-04T12:57:33,964+0000 [pulsar-web-38-6] ERROR org.apache.pulsar.broker.admin.impl.PersistentTopicsBase - [null] Failed to offload status on topic <persistent://offLoadTenant/offloading/test>
java.util.concurrent.CompletionException: javax.ws.rs.WebApplicationException: HTTP 307 Temporary Redirect
        at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315) ~[?:?]
        at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:320) ~[?:?]
        at java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:994) ~[?:?]
        at java.util.concurrent.CompletableFuture.uniExceptionallyStage(CompletableFuture.java:1008) ~[?:?]
        at java.util.concurrent.CompletableFuture.exceptionally(CompletableFuture.java:2364) ~[?:?]
        at org.apache.pulsar.broker.web.PulsarWebResource.validateTopicOwnershipAsync(PulsarWebResource.java:797) ~[org.apache.pulsar-pulsar-broker-3.0.4.jar:3.0.4]
        at org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.internalOffloadStatus(PersistentTopicsBase.java:4392) ~[org.apache.pulsar-pulsar-broker-3.0.4.jar:3.0.4]
        at org.apache.pulsar.broker.admin.v2.PersistentTopics.offloadStatus(PersistentTopics.java:3281) ~[org.apache.pulsar-pulsar-broker-3.0.4.jar:3.0.4]
        at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
        at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77) ~[?:?]
        at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:?]
        at java.lang.reflect.Method.invoke(Method.java:568) ~[?:?]
        at org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory.lambda$static$0(ResourceMethodInvocationHandlerFactory.java:52) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher$1.run(AbstractJavaResourceMethodDispatcher.java:124) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.invoke(AbstractJavaResourceMethodDispatcher.java:167) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.server.model.internal.JavaResourceMethodDispatcherProvider$VoidOutInvoker.doDispatch(JavaResourceMethodDispatcherProvider.java:159) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.dispatch(AbstractJavaResourceMethodDispatcher.java:79) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.server.model.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:475) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:397) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:81) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.server.ServerRuntime$1.run(ServerRuntime.java:255) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]
        at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]
        at org.glassfish.jersey.internal.Errors.process(Errors.java:292) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]
        at org.glassfish.jersey.internal.Errors.process(Errors.java:274) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]
        at org.glassfish.jersey.internal.Errors.process(Errors.java:244) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]
        at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:265) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]
        at org.glassfish.jersey.server.ServerRuntime.process(ServerRuntime.java:234) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.server.ApplicationHandler.handle(ApplicationHandler.java:680) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]
        at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:394) ~[org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar:?]
        at org.glassfish.jersey.servlet.WebComponent.service(WebComponent.java:346) ~[org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar:?]
        at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:366) ~[org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar:?]
        at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:319) ~[org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar:?]
        at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:205) ~[org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar:?]
        at org.eclipse.jetty.servlet.ServletHolder.handle(ServletHolder.java:799) ~[org.eclipse.jetty-jetty-servlet-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.servlet.ServletHandler$ChainEnd.doFilter(ServletHandler.java:1656) ~[org.eclipse.jetty-jetty-servlet-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.servlets.QoSFilter.doFilter(QoSFilter.java:202) ~[org.eclipse.jetty-jetty-servlets-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193) ~[org.eclipse.jetty-jetty-servlet-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1626) ~[org.eclipse.jetty-jetty-servlet-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:552) ~[org.eclipse.jetty-jetty-servlet-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1624) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1440) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:188) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:505) ~[org.eclipse.jetty-jetty-servlet-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1594) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:186) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1355) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.ContextHandlerCollection.handle(ContextHandlerCollection.java:234) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:146) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.StatisticsHandler.handle(StatisticsHandler.java:181) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.Server.handle(Server.java:516) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.HttpChannel.lambda$handle$1(HttpChannel.java:487) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.HttpChannel.dispatch(HttpChannel.java:732) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:479) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:277) ~[org.eclipse.jetty-jetty-server-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311) ~[org.eclipse.jetty-jetty-io-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105) ~[org.eclipse.jetty-jetty-io-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104) ~[org.eclipse.jetty-jetty-io-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.runTask(EatWhatYouKill.java:338) ~[org.eclipse.jetty-jetty-util-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:315) ~[org.eclipse.jetty-jetty-util-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.tryProduce(EatWhatYouKill.java:173) ~[org.eclipse.jetty-jetty-util-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.run(EatWhatYouKill.java:131) ~[org.eclipse.jetty-jetty-util-9.4.54.v20240208.jar:9.4.54.v20240208]
        at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:409) ~[org.eclipse.jetty-jetty-util-9.4.54.v20240208.jar:9.4.54.v20240208]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.netty-netty-common-4.1.105.Final.jar:4.1.105.Final]
        at java.lang.Thread.run(Thread.java:840) ~[?:?]
Caused by: javax.ws.rs.WebApplicationException: HTTP 307 Temporary Redirect
        at org.apache.pulsar.broker.web.PulsarWebResource.lambda$validateTopicOwnershipAsync$24(PulsarWebResource.java:795) ~[org.apache.pulsar-pulsar-broker-3.0.4.jar:3.0.4]
        at java.util.concurrent.CompletableFuture.uniAcceptNow(CompletableFuture.java:757) ~[?:?]
        at java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:735) ~[?:?]
        at java.util.concurrent.CompletableFuture.thenAccept(CompletableFuture.java:2182) ~[?:?]
        at org.apache.pulsar.broker.web.PulsarWebResource.validateTopicOwnershipAsync(PulsarWebResource.java:779) ~[org.apache.pulsar-pulsar-broker-3.0.4.jar:3.0.4]
        ... 65 more
This error message can be found on the broker-0
d
Was that with or without 
-w
?
s
That was with an -w
d
First off, thanks for trying that. Secondly, this is obviously an error; the command shouldn’t return a “success” message.
Can you create an issue in the Apache Pulsar project for this?
s
Yes, I will do this
🙏 1
Open in Slack
PreviousNext
Powered by