[Improve] replace admin calls with client or consumer calls in PulsarHelper

[nlu90]

(If this PR fixes a github issue, please add Fixes #<xyz>.)

Fixes #

(or if this PR is one task of a github issue, please add Master Issue: #<xyz> to link to the master issue.)

Master Issue: #113

Motivation

Reduce the hard dependency on PulsarAadmin as much as possible

Modifications

Replace all admin calls with the equivalent client or consumer calls:

admin call	client or consumer call
admin.topics().createSubscription(...)	consumer.subscribe()
admin.topics().resetCursor(...)	consumer.seek()
admin.topics().deleteSubscription(...)	consumer.unsubscribe()
admin.schemas().getSchemaInfo(...)	clientImpl.getSchema(...)
admin.topics().getLastMessageId(...)	consumer.getLastMessageId
admin.topics().getPartitionedTopicMetadata(...)	clientImpl.getPartitionedTopicMetadata(...)
admin.topics().getList(dest.getNamespace)	client.getLookup.getTopicsUnderNamespace(...)

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

With example applications using the connector, we can observe the following logs from spark driver:

-------------------------------------------
Batch: 4
-------------------------------------------
+--------------------+-----+--------------------+--------------------+--------------------+-----------+-------------------+
|               value|__key|             __topic|         __messageId|       __publishTime|__eventTime|__messageProperties|
+--------------------+-----+--------------------+--------------------+--------------------+-----------+-------------------+
|[48 65 6C 6C 6F 2...| null|persistent://publ...|[08 38 10 50 20 0...|2023-03-28 16:07:...|       null|                 {}|
|[48 65 6C 6C 6F 2...| null|persistent://publ...|[08 38 10 51 20 0...|2023-03-28 16:07:...|       null|                 {}|
|[48 65 6C 6C 6F 2...| null|persistent://publ...|[08 38 10 52 20 0...|2023-03-28 16:07:...|       null|                 {}|
|[48 65 6C 6C 6F 2...| null|persistent://publ...|[08 38 10 53 20 0...|2023-03-28 16:07:...|       null|                 {}|
|[48 65 6C 6C 6F 2...| null|persistent://publ...|[08 38 10 54 20 0...|2023-03-28 16:07:...|       null|                 {}|
|[48 65 6C 6C 6F 2...| null|persistent://publ...|[08 38 10 55 20 0...|2023-03-28 16:07:...|       null|                 {}|
|[48 65 6C 6C 6F 2...| null|persistent://publ...|[08 38 10 56 20 0...|2023-03-28 16:07:...|       null|                 {}|
|[48 65 6C 6C 6F 2...| null|persistent://publ...|[08 38 10 57 20 0...|2023-03-28 16:07:...|       null|                 {}|
|[48 65 6C 6C 6F 2...| null|persistent://publ...|[08 38 10 58 20 0...|2023-03-28 16:07:...|       null|                 {}|
|[48 65 6C 6C 6F 2...| null|persistent://publ...|[08 38 10 59 20 0...|2023-03-28 16:07:...|       null|                 {}|
+--------------------+-----+--------------------+--------------------+--------------------+-----------+-------------------+

23/03/28 16:07:30 INFO WriteToDataSourceV2Exec: Data source write support org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@5d7f2978 committed.
23/03/28 16:07:30 INFO CheckpointFileManager: Writing atomically to file:/tmp/bb486861-bd91-44d6-ae7f-c1a5bd7b9bd2/commits/4 using temp file file:/tmp/bb486861-bd91-44d6-ae7f-c1a5bd7b9bd2/commits/.4.2b8a1aa3-d710-42fd-96ac-349a2326e026.tmp
23/03/28 16:07:30 INFO CheckpointFileManager: Renamed temp file file:/tmp/bb486861-bd91-44d6-ae7f-c1a5bd7b9bd2/commits/.4.2b8a1aa3-d710-42fd-96ac-349a2326e026.tmp to file:/tmp/bb486861-bd91-44d6-ae7f-c1a5bd7b9bd2/commits/4
23/03/28 16:07:30 INFO MicroBatchExecution: Streaming query made progress: {
  "id" : "bfc00ad2-f3df-4b1f-b289-99c4faf596f6",
  "runId" : "77483b6b-627b-445c-96e2-b105d0f2f6ff",
  "name" : null,
  "timestamp" : "2023-03-28T23:07:30.004Z",
  "batchId" : 4,
  "numInputRows" : 10,
  "inputRowsPerSecond" : 0.3333222225925802,
  "processedRowsPerSecond" : 26.52519893899204,
  "durationMs" : {
    "addBatch" : 245,
    "getBatch" : 20,
    "getOffset" : 16,
    "queryPlanning" : 5,
    "triggerExecution" : 376,
    "walCommit" : 47
  },
  "stateOperators" : [ ],
  "sources" : [ {
    "description" : "org.apache.spark.sql.pulsar.PulsarSource@4aaeb58f",
    "startOffset" : {
      "persistent://public/default/spark-test2" : [ 8, 56, 16, 79, 48, 0 ]
    },
    "endOffset" : {
      "persistent://public/default/spark-test2" : [ 8, 56, 16, 89, 48, 0 ]
    },
    "latestOffset" : {
      "persistent://public/default/spark-test2" : [ 8, 56, 16, 89, 48, 0 ]
    },
    "numInputRows" : 10,
    "inputRowsPerSecond" : 0.3333222225925802,
    "processedRowsPerSecond" : 26.52519893899204
  } ],
  "sink" : {
    "description" : "org.apache.spark.sql.execution.streaming.ConsoleTable$@606bc1b6",
    "numOutputRows" : 10
  }
}

Documentation

Check the box below.

Need to update docs?

• doc-required

  (If you need help on updating docs, create a doc issue)

• no-need-doc

  Internal implementation details

• doc

  (If this PR contains doc changes)

[nlu90]

The code style checker identify an wrong error:

I'll ignore it for now.