[Bug] Return value of getPartitionedStats doesn't contain subscription type

[sigikjohn-vargheese]

Search before asking

• I searched in the issues and found nothing similar.

Version

Pulsar version - 2.10.3
OS - Ubuntu Linux

Minimal reproduce step

I am using Pulsar admin client's Java version and creates only partitioned topics. I have created subscriptions also using Java admin api client. In order to get all the details of the topic, I am using PartitionedStats (getTopics().getPartitionedStats()). The subscription section in the response doesn't return "type". Subscription type is returned only for TopicStats (getTopics().getStats()). I would like to get all details using PartitionedStats instead of adding a call for TopicStats.

What did you expect to see?

Subscription type returned in getTopics().getPartitionedStats() call.

What did you see instead?

Subscription type is not returned in getTopics().getPartitionedStats() call.

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[tisonkun]

Hi @sigikjohn-vargheese! The interface PartitionedTopicStats which is returned by getPartitionedStats is extends TopicStats which is returned by getStats.

So any field you can get by the returned value of getStats should be able to fetch via the return value of getPartitionedStats.

If it doesn't satisfy your use case, please provide a reproduce minimal repo or at least a code snippet so that I can understand your usage clearly.

[sigikjohn-vargheese]

@tisonkun Here is pared down snippet. I have a service which connects to Pulsar admin API to create Partitioned topics and then get PartitionedTopicStats. Details from PartitionedTopicStats are then mapped to my POJO. I have a condition to check whether subscription type is empty. From PartitionedTopicStats it always comes back empty because that field is missing in the Admin API response. If I replace PartitionedTopicStats with TopicStats, I can see that the type is not empty.

Please note that this behavior can be verified using CLI as well:

pulsar-admin topics partitioned-stats "topicUri" #--> missing subscription type
pulsar-admin topics stats "partitionedTopicUri"  #--> has subscription type

// Interface to connect to pulsar admin Java client
public interface PulsarAdminService {
    PulsarAdmin getPulsarAdmin();
    Topics getTopics();
    CustomTopic createPartitionedTopic(String namespace, String topicPath, String createdBy, String description, int numPartitions);
}

// Service to connect to pulsar admin Java client
@Service
public class PulsarAdminServiceImpl implements PulsarAdminService {

	private PulsarAdmin pulsarAdmin;

	@Override
    	public Topics getTopics() {
            return getPulsarAdmin().topics();
    	}

     	@Override
    	public CustomTopic createPartitionedTopic(String namespace, String topicPath, String createdBy, String description, int numPartitions) {
            Map<String, String> properties = new HashMap<>();
            properties.put(CREATED_ON, new SimpleDateFormat(DATE_FORMAT).format(new Date()));
            properties.put(DESCRIPTION, description);
            properties.put(CREATED_BY, createdBy);
	    try {
                getTopics().createPartitionedTopic(topicPath, numPartitions, properties);

                PartitionedTopicStats pts = getTopics().getPartitionedStats(topicPath, false);
		return buildTopic(namespace, pts); 
	    } catch (PulsarAdminException.ConflictException e) {
                LOG.warn("Topic already exists", e);
            } 
      	}

	@Override
    	public PulsarAdmin getPulsarAdmin() {
            if(pulsarAdmin == null){
            	pulsarAdmin = createPulsarAdmin();
            }
            return pulsarAdmin;
        }

	private PulsarAdmin createPulsarAdmin() {
            try {
            	PulsarAdminBuilder pulsarAdminBuilder = PulsarAdmin.builder();
            	pulsarAdminBuilder.serviceHttpUrl("http://localhost:8080");
            	pulsarAdminBuilder.authentication(AuthenticationFactory.token("real JWT goes here"));
            	pulsarAdminBuilder.allowTlsInsecureConnection(false);
            	pulsarAdminBuilder.tlsTrustCertsFilePath("some path");
            	pulsarAdminBuilder.enableTlsHostnameVerification(false);
            	return pulsarAdminBuilder.build();
            } catch (RuntimeException | PulsarClientException e) {
            	LOG.error("Failed to create Pulsar Admin instance.", e);
            }
    	}

	private CustomTopic buildTopic(String namespace, PartitionedTopicStats pts) {
	    CustomTopic topic = new CustomTopic();
	    Map<String, String> topicProps = partitionedTopicStats.getMetadata().properties;
	    topic.setName(topicName);
	    topic.setNamespace(namespace);
            topic.setPartitions(partitionedTopicStats.getMetadata().partitions);
            topic.setPersistent(true);
	    topic.setDescription(topicProps.get("description"));
            topic.setCreatedBy(topicProps.get("createdBy"));
	    List<CustomSubscription> subscriptionList = new ArrayList<>();
	    for (Map.Entry<String, ? extends SubscriptionStats> entry : pts.getSubscriptions().entrySet()) {
            	CustomSubscription subscription = new CustomSubscription();
            	subscription.setName(entry.getKey());
		SubscriptionStats subscriptionStats = entry.getValue();
		if (null != subscriptionStats.getType() && !subscriptionStats.getType().trim().isEmpty()) { // DEFECT - type is missing
		    subscription.setType(subscriptionStats.getType()); 
		}
                subscriptionList.add(subscription);
            }
	    topic.setSubscriptions(subscriptionList);
	    return topic;
	}
}

// Custom POJOs
public class CustomTopic implements Serializable {

    public static final String PROP_NAME = "name";
    public static final String PROP_NAMESPACE = "namespace";
    public static final String PROP_PERSISTENT = "persistent";
    public static final String PROP_SUBSCRIPTIONS = "subscriptions";
    public static final String PROP_DESCRIPTION = "description";
    public static final String PROP_CREATED_BY = "createdBy";
    public static final String PROP_PARTITIONS = "partitions";

    @NotNull
    private String name;
    @NotNull
    private String namespace;
    @NotNull
    private Boolean persistent;
    private List<CustomSubscription> subscriptions;
    private String description;
    private String createdBy;
    @NotNull
    private Integer partitions;

    public CustomTopic() {
    }

    public CustomTopic(@NotNull String name, @NotNull String namespace, @NotNull Boolean persistent, @NotNull Integer partitions) {
        this.name = name;
        this.environment = environment;
        this.persistent = persistent;
        this.partitions = partitions;
    }

    @NotNull
    public String getName() {
        return this.name;
    }

    public void setName(@NotNull String name) {
        this.name = name;
    }

    @NotNull
    public String getNamespace() {
        return this.namespace;
    }

    public void setNamespace(@NotNull String namespace) {
        this.namespace = namespace;
    }

    @NotNull
    public Boolean isPersistent() {
        return this.persistent;
    }

    public void setPersistent(@NotNull Boolean persistent) {
        this.persistent = persistent;
    }

    public List<CustomSubscription> getSubscriptions() {
        return this.subscriptions;
    }

    public void setSubscriptions(List<CustomSubscription> subscriptions) {
        this.subscriptions = subscriptions;

    public String getDescription() {
        return this.description;
    }

    public void setDescription(String description) {
        this.description = description;
    }

    public String getCreatedBy() {
        return this.createdBy;
    }

    public void setCreatedBy(String createdBy) {
        this.createdBy = createdBy;
    }

    @NotNull
    public Integer getPartitions() {
        return this.partitions;
    }

    public void setPartitions(@NotNull Integer partitions) {
        this.partitions = partitions;
    }
}

public class CustomSubscription implements Serializable {

    public static final String PROP_NAME = "name";
    public static final String PROP_TYPE = "type";

    @NotNull
    private String name;
    private String type;

    public CustomSubscription() {
    }

    public CustomSubscription(@NotNull String name) {
        this.name = name;
        this.topic = topic;
    }

    @NotNull
    public String getName() {
        return this.name;
    }

    public void setName(@NotNull String name) {
        this.name = name;
    }

    public String getType() {
        return this.type;
    }

    public void setType(String type) {
        this.type = type;
    }
}