ViacheslavKlimov
1 year ago
43 changed files with 795 additions and 136 deletions
@ -0,0 +1,73 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.controller; |
||||
|
|
||||
|
import io.swagger.v3.oas.annotations.Parameter; |
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.security.access.prepost.PreAuthorize; |
||||
|
import org.springframework.web.bind.annotation.GetMapping; |
||||
|
import org.springframework.web.bind.annotation.PathVariable; |
||||
|
import org.springframework.web.bind.annotation.RequestMapping; |
||||
|
import org.springframework.web.bind.annotation.RequestParam; |
||||
|
import org.springframework.web.bind.annotation.RestController; |
||||
|
import org.thingsboard.server.common.data.exception.ThingsboardException; |
||||
|
import org.thingsboard.server.common.data.id.JobId; |
||||
|
import org.thingsboard.server.common.data.job.Job; |
||||
|
import org.thingsboard.server.common.data.page.PageData; |
||||
|
import org.thingsboard.server.common.data.page.PageLink; |
||||
|
import org.thingsboard.server.dao.task.JobService; |
||||
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
|
||||
|
import static org.thingsboard.server.controller.ControllerConstants.PAGE_NUMBER_DESCRIPTION; |
||||
|
import static org.thingsboard.server.controller.ControllerConstants.PAGE_SIZE_DESCRIPTION; |
||||
|
import static org.thingsboard.server.controller.ControllerConstants.SORT_ORDER_DESCRIPTION; |
||||
|
import static org.thingsboard.server.controller.ControllerConstants.SORT_PROPERTY_DESCRIPTION; |
||||
|
|
||||
|
@RestController |
||||
|
@TbCoreComponent |
||||
|
@RequestMapping("/api") |
||||
|
@RequiredArgsConstructor |
||||
|
@Slf4j |
||||
|
public class JobController extends BaseController { |
||||
|
|
||||
|
private final JobService jobService; |
||||
|
|
||||
|
@GetMapping("/job/{id}") |
||||
|
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
||||
|
public Job getJobById(@PathVariable UUID id) throws ThingsboardException { |
||||
|
return jobService.findJobById(getTenantId(), new JobId(id)); |
||||
|
} |
||||
|
|
||||
|
@GetMapping("/jobs") |
||||
|
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
||||
|
public PageData<Job> getJobs(@Parameter(description = PAGE_SIZE_DESCRIPTION, required = true) |
||||
|
@RequestParam int pageSize, |
||||
|
@Parameter(description = PAGE_NUMBER_DESCRIPTION, required = true) |
||||
|
@RequestParam int page, |
||||
|
@Parameter(description = "Case-insensitive 'substring' filter based on job's description") |
||||
|
@RequestParam(required = false) String textSearch, |
||||
|
@Parameter(description = SORT_PROPERTY_DESCRIPTION) |
||||
|
@RequestParam(required = false) String sortProperty, |
||||
|
@Parameter(description = SORT_ORDER_DESCRIPTION) |
||||
|
@RequestParam(required = false) String sortOrder) throws ThingsboardException { |
||||
|
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
||||
|
return jobService.findJobsByTenantId(getTenantId(), pageLink); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,64 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.service.job; |
||||
|
|
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.common.data.job.DummyJobConfiguration; |
||||
|
import org.thingsboard.server.common.data.job.DummyTask; |
||||
|
import org.thingsboard.server.common.data.job.Job; |
||||
|
import org.thingsboard.server.common.data.job.JobType; |
||||
|
import org.thingsboard.server.common.data.job.Task; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.function.Consumer; |
||||
|
|
||||
|
@Component |
||||
|
@RequiredArgsConstructor |
||||
|
public class DummyJobProcessor extends JobProcessor { |
||||
|
|
||||
|
@Override |
||||
|
public int process(Job job, Consumer<Task> taskConsumer) { |
||||
|
DummyJobConfiguration configuration = job.getConfiguration(); |
||||
|
for (int number = 1; number <= configuration.getSuccessfulTasksCount(); number++) { |
||||
|
taskConsumer.accept(createTask(job, configuration, number, null)); |
||||
|
} |
||||
|
if (configuration.getErrors() != null) { |
||||
|
for (int number = 1; number <= configuration.getFailedTasksCount(); number++) { |
||||
|
taskConsumer.accept(createTask(job, configuration, number, configuration.getErrors())); |
||||
|
} |
||||
|
} |
||||
|
return configuration.getSuccessfulTasksCount() + configuration.getFailedTasksCount(); |
||||
|
} |
||||
|
|
||||
|
private Task createTask(Job job, DummyJobConfiguration configuration, int number, List<String> errors) { |
||||
|
return DummyTask.builder() |
||||
|
.tenantId(job.getTenantId()) |
||||
|
.jobId(job.getId()) |
||||
|
.key("Task " + number) |
||||
|
.retries(configuration.getRetries()) |
||||
|
.number(number) |
||||
|
.processingTimeMs(configuration.getTaskProcessingTimeMs()) |
||||
|
.errors(errors) |
||||
|
.build(); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public JobType getType() { |
||||
|
return JobType.DUMMY; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,44 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.service.job.task; |
||||
|
|
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.common.data.job.DummyTask; |
||||
|
import org.thingsboard.server.common.data.job.JobType; |
||||
|
import org.thingsboard.server.queue.task.TaskProcessor; |
||||
|
|
||||
|
@Component |
||||
|
@RequiredArgsConstructor |
||||
|
public class DummyTaskProcessor extends TaskProcessor<DummyTask> { |
||||
|
|
||||
|
@Override |
||||
|
protected void process(DummyTask task) throws Exception { |
||||
|
if (task.getProcessingTimeMs() > 0) { |
||||
|
Thread.sleep(task.getProcessingTimeMs()); |
||||
|
} |
||||
|
if (task.getErrors() != null && task.getAttempt() <= task.getErrors().size()) { |
||||
|
String error = task.getErrors().get(task.getAttempt() - 1); |
||||
|
throw new RuntimeException(error); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public JobType getJobType() { |
||||
|
return JobType.DUMMY; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,125 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.service.job; |
||||
|
|
||||
|
import com.fasterxml.jackson.core.type.TypeReference; |
||||
|
import org.junit.After; |
||||
|
import org.junit.Before; |
||||
|
import org.junit.Test; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.test.context.TestPropertySource; |
||||
|
import org.thingsboard.server.common.data.id.JobId; |
||||
|
import org.thingsboard.server.common.data.job.DummyJobConfiguration; |
||||
|
import org.thingsboard.server.common.data.job.Job; |
||||
|
import org.thingsboard.server.common.data.job.JobStatus; |
||||
|
import org.thingsboard.server.common.data.job.JobType; |
||||
|
import org.thingsboard.server.common.data.page.PageData; |
||||
|
import org.thingsboard.server.common.data.page.PageLink; |
||||
|
import org.thingsboard.server.controller.AbstractControllerTest; |
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
@TestPropertySource(properties = { |
||||
|
"queue.tasks.stats.processing_interval_ms=0" |
||||
|
}) |
||||
|
public class JobManagerTest extends AbstractControllerTest { |
||||
|
|
||||
|
@Autowired |
||||
|
private JobManager jobManager; |
||||
|
|
||||
|
@Before |
||||
|
public void setUp() throws Exception { |
||||
|
loginTenantAdmin(); |
||||
|
} |
||||
|
|
||||
|
@After |
||||
|
public void tearDown() throws Exception { |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testSubmitJob_allTasksSuccessful() { |
||||
|
int tasksCount = 5; |
||||
|
JobId jobId = jobManager.submitJob(Job.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.type(JobType.DUMMY) |
||||
|
.key("test-job") |
||||
|
.description("test job") |
||||
|
.configuration(DummyJobConfiguration.builder() |
||||
|
.successfulTasksCount(tasksCount) |
||||
|
.taskProcessingTimeMs(1000) |
||||
|
.build()) |
||||
|
.build()).getId(); |
||||
|
|
||||
|
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
||||
|
Job job = findJobById(jobId); |
||||
|
assertThat(job.getStatus()).isEqualTo(JobStatus.RUNNING); |
||||
|
assertThat(job.getResult().getSuccessfulCount()).isBetween(1, tasksCount - 1); |
||||
|
assertThat(job.getResult().getTotalCount()).isEqualTo(tasksCount); |
||||
|
}); |
||||
|
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
||||
|
Job job = findJobById(jobId); |
||||
|
assertThat(job.getStatus()).isEqualTo(JobStatus.COMPLETED); |
||||
|
assertThat(job.getResult().getSuccessfulCount()).isEqualTo(tasksCount); |
||||
|
assertThat(job.getResult().getFailures()).isEmpty(); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testSubmitJob_someTasksPermanentlyFailed() { |
||||
|
int successfulTasks = 3; |
||||
|
int failedTasks = 2; |
||||
|
JobId jobId = jobManager.submitJob(Job.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.type(JobType.DUMMY) |
||||
|
.key("test-job") |
||||
|
.description("test job") |
||||
|
.configuration(DummyJobConfiguration.builder() |
||||
|
.successfulTasksCount(successfulTasks) |
||||
|
.failedTasksCount(failedTasks) |
||||
|
.errors(List.of("error1", "error2", "error3")) |
||||
|
.retries(2) |
||||
|
.taskProcessingTimeMs(100) |
||||
|
.build()) |
||||
|
.build()).getId(); |
||||
|
|
||||
|
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
||||
|
Job job = findJobById(jobId); |
||||
|
assertThat(job.getStatus()).isEqualTo(JobStatus.FAILED); |
||||
|
assertThat(job.getResult().getSuccessfulCount()).isEqualTo(successfulTasks); |
||||
|
assertThat(job.getResult().getFailedCount()).isEqualTo(failedTasks); |
||||
|
assertThat(job.getResult().getTotalCount()).isEqualTo(successfulTasks + failedTasks); |
||||
|
assertThat(job.getResult().getFailures().get("Task 1")).isEqualTo("error3"); // last error
|
||||
|
assertThat(job.getResult().getFailures().get("Task 2")).isEqualTo("error3"); // last error
|
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
|
||||
|
|
||||
|
private Job findJobById(JobId jobId) throws Exception { |
||||
|
return doGet("/api/job/" + jobId, Job.class); |
||||
|
} |
||||
|
|
||||
|
private List<Job> findJobs() throws Exception { |
||||
|
return doGetTypedWithPageLink("/api/jobs?", new TypeReference<PageData<Job>>() {}, new PageLink(100, 0)).getData(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,42 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.common.data.job; |
||||
|
|
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import lombok.NoArgsConstructor; |
||||
|
|
||||
|
import java.util.List; |
||||
|
|
||||
|
@Data |
||||
|
@AllArgsConstructor |
||||
|
@NoArgsConstructor |
||||
|
@Builder |
||||
|
public class DummyJobConfiguration implements JobConfiguration { |
||||
|
|
||||
|
private long taskProcessingTimeMs; |
||||
|
private int successfulTasksCount; |
||||
|
private int failedTasksCount; |
||||
|
private List<String> errors; |
||||
|
private int retries; |
||||
|
|
||||
|
@Override |
||||
|
public JobType getType() { |
||||
|
return JobType.DUMMY; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,25 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.common.data.job; |
||||
|
|
||||
|
public class DummyJobResult extends JobResult { |
||||
|
|
||||
|
@Override |
||||
|
public JobType getJobType() { |
||||
|
return JobType.DUMMY; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,40 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.common.data.job; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
import lombok.EqualsAndHashCode; |
||||
|
import lombok.NoArgsConstructor; |
||||
|
import lombok.experimental.SuperBuilder; |
||||
|
|
||||
|
import java.util.List; |
||||
|
|
||||
|
@Data |
||||
|
@NoArgsConstructor |
||||
|
@EqualsAndHashCode(callSuper = true) |
||||
|
@SuperBuilder |
||||
|
public class DummyTask extends Task { |
||||
|
|
||||
|
private int number; |
||||
|
private long processingTimeMs; |
||||
|
private List<String> errors; // errors for each attempt
|
||||
|
|
||||
|
@Override |
||||
|
public JobType getJobType() { |
||||
|
return JobType.DUMMY; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,29 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.common.data.job; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
import org.thingsboard.server.common.data.id.JobId; |
||||
|
|
||||
|
import java.util.ArrayList; |
||||
|
import java.util.List; |
||||
|
|
||||
|
@Data |
||||
|
public class JobStats { |
||||
|
private final JobId jobId; |
||||
|
private final List<TaskResult> taskResults = new ArrayList<>(); |
||||
|
private Integer totalTasksCount; |
||||
|
} |
||||
@ -0,0 +1,63 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.queue.task; |
||||
|
|
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.context.annotation.Lazy; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.server.common.data.id.JobId; |
||||
|
import org.thingsboard.server.common.data.job.TaskResult; |
||||
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.JobStatsMsg; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.TaskResultProto; |
||||
|
import org.thingsboard.server.queue.TbQueueCallback; |
||||
|
import org.thingsboard.server.queue.TbQueueProducer; |
||||
|
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
||||
|
import org.thingsboard.server.queue.provider.TbQueueProducerProvider; |
||||
|
|
||||
|
@Lazy |
||||
|
@Service |
||||
|
@Slf4j |
||||
|
@RequiredArgsConstructor |
||||
|
public class JobStatsService { |
||||
|
|
||||
|
private final TbQueueProducerProvider producerProvider; |
||||
|
|
||||
|
public void reportTaskResult(JobId jobId, TaskResult result) { |
||||
|
report(jobId, JobStatsMsg.newBuilder() |
||||
|
.setTaskResult(TaskResultProto.newBuilder() |
||||
|
.setValue(JacksonUtil.toString(result)) |
||||
|
.build())); |
||||
|
} |
||||
|
|
||||
|
public void reportAllTasksSubmitted(JobId jobId, int tasksCount) { |
||||
|
report(jobId, JobStatsMsg.newBuilder() |
||||
|
.setTotalTasksCount(tasksCount)); |
||||
|
} |
||||
|
|
||||
|
private void report(JobId jobId, JobStatsMsg.Builder statsMsg) { |
||||
|
log.info("[{}] Reporting: {}", jobId, statsMsg); |
||||
|
statsMsg.setJobIdMSB(jobId.getId().getMostSignificantBits()) |
||||
|
.setJobIdLSB(jobId.getId().getLeastSignificantBits()); |
||||
|
|
||||
|
TbProtoQueueMsg<JobStatsMsg> msg = new TbProtoQueueMsg<>(jobId.getId(), statsMsg.build()); |
||||
|
TbQueueProducer<TbProtoQueueMsg<JobStatsMsg>> producer = producerProvider.getJobStatsProducer(); |
||||
|
producer.send(TopicPartitionInfo.builder().topic(producer.getDefaultTopic()).build(), msg, TbQueueCallback.EMPTY); |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue