Skip to content

[Functions] Support KEY_BASED batch builder for Java based functions and sources #11706

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lhotari merged 3 commits into from
Aug 25, 2021
Merged

[Functions] Support KEY_BASED batch builder for Java based functions and sources #11706

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f0437b7
[Functions] Support KEY_BASED batch builder for Java based functions …
lhotari Aug 24, 2021
8d5042d
Include batchBuilder in ProducerSpec -> ProducerConfig.ProducerConfig…
lhotari Aug 19, 2021
1a1f10a
Support setting batch builder for sources with "--batch-builder KEY_B…
lhotari Aug 19, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 15 additions & 6 deletions ...ar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import lombok.ToString;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.BatcherBuilder;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.Consumer;
Expand Down Expand Up @@ -146,14 +147,22 @@ public ContextImpl(InstanceConfig config, Logger logger, PulsarClient client,
this.producerBuilder = (ProducerBuilderImpl<?>) client.newProducer().blockIfQueueFull(true).enableBatching(true)
.batchingMaxPublishDelay(1, TimeUnit.MILLISECONDS);
boolean useThreadLocalProducers = false;
if (config.getFunctionDetails().getSink().getProducerSpec() != null) {
if (config.getFunctionDetails().getSink().getProducerSpec().getMaxPendingMessages() != 0) {
this.producerBuilder.maxPendingMessages(config.getFunctionDetails().getSink().getProducerSpec().getMaxPendingMessages());
Function.ProducerSpec producerSpec = config.getFunctionDetails().getSink().getProducerSpec();
if (producerSpec != null) {
if (producerSpec.getMaxPendingMessages() != 0) {
this.producerBuilder.maxPendingMessages(producerSpec.getMaxPendingMessages());
}
if (config.getFunctionDetails().getSink().getProducerSpec().getMaxPendingMessagesAcrossPartitions() != 0) {
this.producerBuilder.maxPendingMessagesAcrossPartitions(config.getFunctionDetails().getSink().getProducerSpec().getMaxPendingMessagesAcrossPartitions());
if (producerSpec.getMaxPendingMessagesAcrossPartitions() != 0) {
this.producerBuilder.maxPendingMessagesAcrossPartitions(producerSpec.getMaxPendingMessagesAcrossPartitions());
}
useThreadLocalProducers = config.getFunctionDetails().getSink().getProducerSpec().getUseThreadLocalProducers();
if (producerSpec.getBatchBuilder() != null) {
if (producerSpec.getBatchBuilder().equals("KEY_BASED")) {
this.producerBuilder.batcherBuilder(BatcherBuilder.KEY_BASED);
} else {
this.producerBuilder.batcherBuilder(BatcherBuilder.DEFAULT);
}
}
useThreadLocalProducers = producerSpec.getUseThreadLocalProducers();
}
if (useThreadLocalProducers) {
tlPublishProducers = new ThreadLocal<>();
Expand Down
1 change: 1 addition & 0 deletions ...ons/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -782,6 +782,7 @@ private void setupOutput(ContextImpl contextImpl) throws Exception {
ProducerConfig.ProducerConfigBuilder builder = ProducerConfig.builder()
.maxPendingMessages(conf.getMaxPendingMessages())
.maxPendingMessagesAcrossPartitions(conf.getMaxPendingMessagesAcrossPartitions())
.batchBuilder(conf.getBatchBuilder())
.useThreadLocalProducers(conf.getUseThreadLocalProducers())
.cryptoConfig(CryptoUtils.convertFromSpec(conf.getCryptoSpec()));
pulsarSinkConfig.setProducerConfig(builder.build());
Expand Down
7 changes: 7 additions & 0 deletions ...ar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,13 @@ public static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSource
sinkSpecBuilder.setProducerSpec(pbldr.build());
}

if (sourceConfig.getBatchBuilder() != null) {
Function.ProducerSpec.Builder builder = sinkSpecBuilder.getProducerSpec() != null
? sinkSpecBuilder.getProducerSpec().toBuilder()
: Function.ProducerSpec.newBuilder();
sinkSpecBuilder.setProducerSpec(builder.setBatchBuilder(sourceConfig.getBatchBuilder()).build());
}

sinkSpecBuilder.setForwardSourceMessageProperty(true);

functionDetailsBuilder.setSink(sinkSpecBuilder);
Expand Down
31 changes: 31 additions & 0 deletions ...unctions/utils/src/test/java/org/apache/pulsar/functions/utils/SourceConfigUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -331,6 +331,37 @@ public void testValidateConfig() throws IOException {
assertTrue(e.getMessage().contains("Could not validate source config: Field 'configParameter' cannot be null!"));
}

@Test
public void testSupportsBatchBuilderWhenProducerConfigIsNull() {
SourceConfig sourceConfig = createSourceConfig();
sourceConfig.setProducerConfig(null);
sourceConfig.setBatchBuilder("KEY_BASED");
Function.FunctionDetails functionDetails =
SourceConfigUtils.convert(sourceConfig, new SourceConfigUtils.ExtractedSourceDetails(null, null));
assertEquals(functionDetails.getSink().getProducerSpec().getBatchBuilder(), "KEY_BASED");
}

@Test
public void testSupportsBatchBuilderWhenProducerConfigExists() {
SourceConfig sourceConfig = createSourceConfig();
sourceConfig.setBatchBuilder("KEY_BASED");
sourceConfig.getProducerConfig().setMaxPendingMessages(123456);
Function.FunctionDetails functionDetails =
SourceConfigUtils.convert(sourceConfig, new SourceConfigUtils.ExtractedSourceDetails(null, null));
assertEquals(functionDetails.getSink().getProducerSpec().getBatchBuilder(), "KEY_BASED");
assertEquals(functionDetails.getSink().getProducerSpec().getMaxPendingMessages(), 123456);
}

@Test
public void testSupportsBatchBuilderDefinedInProducerConfigWhenTopLevelBatchBuilderIsUndefined() {
SourceConfig sourceConfig = createSourceConfig();
sourceConfig.setBatchBuilder(null);
sourceConfig.getProducerConfig().setBatchBuilder("KEY_BASED");
Function.FunctionDetails functionDetails =
SourceConfigUtils.convert(sourceConfig, new SourceConfigUtils.ExtractedSourceDetails(null, null));
assertEquals(functionDetails.getSink().getProducerSpec().getBatchBuilder(), "KEY_BASED");
}

private SourceConfig createSourceConfigWithBatch() {
SourceConfig sourceConfig = createSourceConfig();
BatchSourceConfig batchSourceConfig = createBatchSourceConfig();
Expand Down