Simplify creation of functions namespace
@@ -24,6 +24,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.conf.InternalConfigurationData;
+import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.functions.worker.rest.WorkerServer;
@@ -101,7 +102,10 @@ public class Worker extends AbstractService {
if (e.getStatusCode() == Response.Status.NOT_FOUND.getStatusCode()) {
// if not found than create
try {
- admin.namespaces().createNamespace(workerConfig.getPulsarFunctionsNamespace());
+ Policies policies = new Policies();
+ policies.retention_policies = new RetentionPolicies(-1, -1);
+ admin.namespaces().createNamespace(workerConfig.getPulsarFunctionsNamespace(),
+ policies);
} catch (PulsarAdminException e1) {
// prevent race condition with other workers starting up
if (e1.getStatusCode() != Response.Status.CONFLICT.getStatusCode()) {
@@ -110,14 +114,6 @@ public class Worker extends AbstractService {
throw e1;
}
}
- try {
- admin.namespaces().setRetention(
- workerConfig.getPulsarFunctionsNamespace(),
- new RetentionPolicies(Integer.MAX_VALUE, Integer.MAX_VALUE));
- } catch (PulsarAdminException e1) {
- log.error("Failed to set retention policy for pulsar functions namespace", e);
- throw new RuntimeException(e1);
- }
} else {
log.error("Failed to get retention policy for pulsar function namespace {}",
workerConfig.getPulsarFunctionsNamespace(), e);
