[Issue 10445][pulsar-io] Exposed SubscriptionType in the SinkContext

[dlg99]

Fixes #10445

Motivation

SinkContext should expose Subscription type to the Sink
More context: #9927 (comment)

Needed for #9927

Modifications

Added getSubscriptionType() to the SinkContext interface and ContextImpl

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing unit tests

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• The public API: yes

New method added to the interface that is a public API. Default method implementation is provided.

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? JavaDocs

[eolivelli]

overall is good.

I would also add a (unit, non integration) test in order to ensure that the Sink receives the value.

[eolivelli]

Lgtm

[dlg99]

rebased on current master

[sijie]

before merging this pull request, we should get stakeholders like @jerrypeng and @srkukarni to confirm if this API works for them.

[dlg99]

@jerrypeng @srkukarni Do you have any feedback on this? (I cannot add you to reviewers, no permissions)

[eolivelli]

@jerrypeng @srkukarni ping.
this patch is blocking another bigger work.
can you please take a quick look ?

[jerrypeng]

I don't see any red flags in this PR but I also don't understand adding this functionality will be useful.

@dlg99 based the comments

#9927 (comment)

It seem like the subscription type should be checked at submission time and be rejected if a wrong subscription type is used.

If we are checking the subscription type during the sink open, what are we going to do if the subscription type is wrong? Just go into a crash loop?

[dlg99]

If we are checking the subscription type during the sink open, what are we going to do if the subscription type is wrong? Just go into a crash loop?

@jerrypeng This is not different from what's happening in other connectors now. I agree we can consider adding something to handle it better, but this is out of the scope of this PR.

pulsar/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java

Lines 84 to 91 in 5a1264d

	if (kafkaSinkConfig.getBatchSize() <= 0) {
	throw new IllegalArgumentException("Invalid Kafka Producer batchSize : "
	+ kafkaSinkConfig.getBatchSize());
	}
	if (kafkaSinkConfig.getMaxRequestSize() <= 0) {
	throw new IllegalArgumentException("Invalid Kafka Producer maxRequestSize : "
	+ kafkaSinkConfig.getMaxRequestSize());
	}

pulsar/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java

Lines 159 to 163 in 5a1264d

	checkArgument(isNotBlank(kinesisSinkConfig.getAwsKinesisStreamName()), "empty kinesis-stream name");
	checkArgument(isNotBlank(kinesisSinkConfig.getAwsEndpoint()) ||
	isNotBlank(kinesisSinkConfig.getAwsRegion()),
	"Either the aws-end-point or aws-region must be set");
	checkArgument(isNotBlank(kinesisSinkConfig.getAwsCredentialPluginParam()), "empty aws-credential param");

etc

[jerrypeng]

I am not a fan of adding new interfaces to support a not well thought out behavior that might be changed in the future but I don't see a major problem with this as well.

[eolivelli]

Thank you @jerrypeng

@sijie you left "request changes" status.
Do you agree to merge this patch ?