Skip to content

Commit

Permalink
Set function resources properly (apache#2841)
Browse files Browse the repository at this point in the history 
* Set function resources properly

* Added unittests
  • Loading branch information
@srkukarni @jerrypeng
srkukarni authored and jerrypeng committed Oct 25, 2018
1 parent 1026edc commit a9a7f30
Show file tree
Hide file tree
Showing 2 changed files with 203 additions and 1 deletion.
187 changes: 187 additions & 0 deletions pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -811,6 +811,193 @@ public void TestCreateClassName() throws Exception {
testValidateFunctionsConfigs(correctArgs, incorrectArgs, "User class must be in class path");
}

@Test
public void testCreateFunctionWithCpu() throws Exception {
String fnName = TEST_NAME + "-function";
String inputTopicName = TEST_NAME + "-input-topic";
String outputTopicName = TEST_NAME + "-output-topic";

final String url = "file:" + JAR_NAME;
cmd.run(new String[] {
"create",
"--name", fnName,
"--inputs", inputTopicName,
"--output", outputTopicName,
"--jar", url,
"--tenant", "sample",
"--namespace", "ns1",
"--className", DummyFunction.class.getName(),
"--cpu", "5.0"
});

CreateFunction creater = cmd.getCreater();

assertEquals(fnName, creater.getFunctionName());
assertEquals(inputTopicName, creater.getInputs());
assertEquals(outputTopicName, creater.getOutput());
assertEquals(creater.getFunctionConfig().getResources().getCpu(), 5.0);
// Disk/Ram should be default
assertEquals(creater.getFunctionConfig().getResources().getRam(), new Long(1073741824l));
assertEquals(creater.getFunctionConfig().getResources().getDisk(), new Long(10737418240l));
verify(functions, times(1)).createFunctionWithUrl(any(FunctionConfig.class), anyString());
}

@Test
public void testCreateFunctionWithRam() throws Exception {
String fnName = TEST_NAME + "-function";
String inputTopicName = TEST_NAME + "-input-topic";
String outputTopicName = TEST_NAME + "-output-topic";

final String url = "file:" + JAR_NAME;
cmd.run(new String[] {
"create",
"--name", fnName,
"--inputs", inputTopicName,
"--output", outputTopicName,
"--jar", url,
"--tenant", "sample",
"--namespace", "ns1",
"--className", DummyFunction.class.getName(),
"--ram", "5656565656"
});

CreateFunction creater = cmd.getCreater();

assertEquals(fnName, creater.getFunctionName());
assertEquals(inputTopicName, creater.getInputs());
assertEquals(outputTopicName, creater.getOutput());
assertEquals(creater.getFunctionConfig().getResources().getRam(), new Long(5656565656l));
// cpu/disk should be default
assertEquals(creater.getFunctionConfig().getResources().getCpu(), 1.0);
assertEquals(creater.getFunctionConfig().getResources().getDisk(), new Long(10737418240l));
verify(functions, times(1)).createFunctionWithUrl(any(FunctionConfig.class), anyString());
}

@Test
public void testCreateFunctionWithDisk() throws Exception {
String fnName = TEST_NAME + "-function";
String inputTopicName = TEST_NAME + "-input-topic";
String outputTopicName = TEST_NAME + "-output-topic";

final String url = "file:" + JAR_NAME;
cmd.run(new String[] {
"create",
"--name", fnName,
"--inputs", inputTopicName,
"--output", outputTopicName,
"--jar", url,
"--tenant", "sample",
"--namespace", "ns1",
"--className", DummyFunction.class.getName(),
"--disk", "8080808080808080"
});

CreateFunction creater = cmd.getCreater();

assertEquals(fnName, creater.getFunctionName());
assertEquals(inputTopicName, creater.getInputs());
assertEquals(outputTopicName, creater.getOutput());
assertEquals(creater.getFunctionConfig().getResources().getDisk(), new Long(8080808080808080l));
// cpu/Ram should be default
assertEquals(creater.getFunctionConfig().getResources().getRam(), new Long(1073741824l));
assertEquals(creater.getFunctionConfig().getResources().getCpu(), 1.0);
verify(functions, times(1)).createFunctionWithUrl(any(FunctionConfig.class), anyString());
}


@Test
public void testUpdateFunctionWithCpu() throws Exception {
String fnName = TEST_NAME + "-function";
String inputTopicName = TEST_NAME + "-input-topic";
String outputTopicName = TEST_NAME + "-output-topic";

final String url = "file:" + JAR_NAME;
cmd.run(new String[] {
"update",
"--name", fnName,
"--inputs", inputTopicName,
"--output", outputTopicName,
"--jar", url,
"--tenant", "sample",
"--namespace", "ns1",
"--className", DummyFunction.class.getName(),
"--cpu", "5.0"
});

UpdateFunction updater = cmd.getUpdater();

assertEquals(fnName, updater.getFunctionName());
assertEquals(inputTopicName, updater.getInputs());
assertEquals(outputTopicName, updater.getOutput());
assertEquals(updater.getFunctionConfig().getResources().getCpu(), 5.0);
// Disk/Ram should be default
assertEquals(updater.getFunctionConfig().getResources().getRam(), new Long(1073741824l));
assertEquals(updater.getFunctionConfig().getResources().getDisk(), new Long(10737418240l));
verify(functions, times(1)).updateFunctionWithUrl(any(FunctionConfig.class), anyString());
}

@Test
public void testUpdateFunctionWithRam() throws Exception {
String fnName = TEST_NAME + "-function";
String inputTopicName = TEST_NAME + "-input-topic";
String outputTopicName = TEST_NAME + "-output-topic";

final String url = "file:" + JAR_NAME;
cmd.run(new String[] {
"update",
"--name", fnName,
"--inputs", inputTopicName,
"--output", outputTopicName,
"--jar", url,
"--tenant", "sample",
"--namespace", "ns1",
"--className", DummyFunction.class.getName(),
"--ram", "5656565656"
});

UpdateFunction updater = cmd.getUpdater();

assertEquals(fnName, updater.getFunctionName());
assertEquals(inputTopicName, updater.getInputs());
assertEquals(outputTopicName, updater.getOutput());
assertEquals(updater.getFunctionConfig().getResources().getRam(), new Long(5656565656l));
// cpu/disk should be default
assertEquals(updater.getFunctionConfig().getResources().getCpu(), 1.0);
assertEquals(updater.getFunctionConfig().getResources().getDisk(), new Long(10737418240l));
verify(functions, times(1)).updateFunctionWithUrl(any(FunctionConfig.class), anyString());
}

@Test
public void testUpdateFunctionWithDisk() throws Exception {
String fnName = TEST_NAME + "-function";
String inputTopicName = TEST_NAME + "-input-topic";
String outputTopicName = TEST_NAME + "-output-topic";

final String url = "file:" + JAR_NAME;
cmd.run(new String[] {
"update",
"--name", fnName,
"--inputs", inputTopicName,
"--output", outputTopicName,
"--jar", url,
"--tenant", "sample",
"--namespace", "ns1",
"--className", DummyFunction.class.getName(),
"--disk", "8080808080808080"
});

UpdateFunction updater = cmd.getUpdater();

assertEquals(fnName, updater.getFunctionName());
assertEquals(inputTopicName, updater.getInputs());
assertEquals(outputTopicName, updater.getOutput());
assertEquals(updater.getFunctionConfig().getResources().getDisk(), new Long(8080808080808080l));
// cpu/Ram should be default
assertEquals(updater.getFunctionConfig().getResources().getRam(), new Long(1073741824l));
assertEquals(updater.getFunctionConfig().getResources().getCpu(), 1.0);
verify(functions, times(1)).updateFunctionWithUrl(any(FunctionConfig.class), anyString());
}

@Test
public void TestCreateSameInOutTopic() throws Exception {

Expand Down
17 changes: 16 additions & 1 deletion pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -414,7 +414,22 @@ void processArguments() throws Exception {
functionConfig.setParallelism(parallelism);
}

functionConfig.setResources(new Resources(cpu, ram, disk));
Resources resources = functionConfig.getResources();
if (resources == null) {
resources = new Resources();
}
if (cpu != null) {
resources.setCpu(cpu);
}

if (ram != null) {
resources.setRam(ram);
}

if (disk != null) {
resources.setDisk(disk);
}
functionConfig.setResources(resources);

if (timeoutMs != null) {
functionConfig.setTimeoutMs(timeoutMs);
Expand Down

0 comments on commit a9a7f30

Please sign in to comment.