Helper method to publish event with correlationId

@param jobInstance
@param topicToPublish
@param correlationId
337774
private void publishEvent(WSJobInstance jobInst, String topicToPublish, String correlationId) {
        if (eventsPublisher != null) {
            eventsPublisher.publishJobInstanceEvent(jobInst, topicToPublish, correlationId);
        }
    }
1008130
@Override
  public <T> T publish(Class<T> beanType, Object id, Transaction transaction) {
    methodCalls.add(MethodCall.of(PUBLISH).with("beanType", beanType).with("id", id));
    return !persistPublish ? null : publish.publish(beanType, id, transaction);
  }
1341194
public void publishInternal(Object message) {
        topicStats.incrementPublishes();
        topicService.publishMessage(name, message, multithreaded);
    }
1108643
public void publishJob(CmsObject cms, CmsDbContext dbc, CmsPublishList publishList, I_CmsReport report)
    throws CmsException {

        try {
            // check state and lock
            List<CmsResource> allResources = new ArrayList<CmsResource>(publishList.getFolderList());
            allResources.addAll(publishList.getDeletedFolderList());
            allResources.addAll(publishList.getFileList());
            Iterator<CmsResource> itResources = allResources.iterator();
            while (itResources.hasNext()) {
                CmsResource resource = itResources.next();
                try {
                    resource = readResource(dbc, resource.getStructureId(), CmsResourceFilter.ALL);
                } catch (CmsVfsResourceNotFoundException e) {
                    continue;
                }
                if (resource.getState().isUnchanged()) {
                    // remove files that were published by a concurrent job
                    if (LOG.isDebugEnabled()) {
                        LOG.debug(
                            Messages.get().getBundle().key(
                                Messages.RPT_PUBLISH_REMOVED_RESOURCE_1,
                                dbc.removeSiteRoot(resource.getRootPath())));
                    }
                    publishList.remove(resource);
                    unlockResource(dbc, resource, true, true);
                    continue;
                }
                CmsLock lock = m_lockManager.getLock(dbc, resource, false);
                if (!lock.getSystemLock().isPublish()) {
                    // remove files that are not locked for publishing
                    if (LOG.isDebugEnabled()) {
                        LOG.debug(
                            Messages.get().getBundle().key(
                                Messages.RPT_PUBLISH_REMOVED_RESOURCE_1,
                                dbc.removeSiteRoot(resource.getRootPath())));
                    }
                    publishList.remove(resource);
                    continue;
                }
            }

            CmsProject onlineProject = readProject(dbc, CmsProject.ONLINE_PROJECT_ID);

            // clear the cache
            m_monitor.clearCache();

            int publishTag = getNextPublishTag(dbc);
            getProjectDriver(dbc).publishProject(dbc, report, onlineProject, publishList, publishTag);

            // iterate the initialized module action instances
            Iterator<String> i = OpenCms.getModuleManager().getModuleNames().iterator();
            while (i.hasNext()) {
                CmsModule module = OpenCms.getModuleManager().getModule(i.next());
                if ((module != null) && (module.getActionInstance() != null)) {
                    module.getActionInstance().publishProject(cms, publishList, publishTag, report);
                }
            }

            boolean temporaryProject = (cms.getRequestContext().getCurrentProject().getType() == CmsProject.PROJECT_TYPE_TEMPORARY);
            // the project was stored in the history tables for history
            // it will be deleted if the project_flag is PROJECT_TYPE_TEMPORARY
            if ((temporaryProject) && (!publishList.isDirectPublish())) {
                try {
                    getProjectDriver(dbc).deleteProject(dbc, dbc.currentProject());
                } catch (CmsException e) {
                    LOG.error(
                        Messages.get().getBundle().key(
                            Messages.LOG_DELETE_TEMP_PROJECT_FAILED_1,
                            cms.getRequestContext().getCurrentProject().getName()));
                }
                // if project was temporary set context to online project
                cms.getRequestContext().setCurrentProject(onlineProject);
            }
        } finally {
            // clear the cache again
            m_monitor.clearCache();
        }
    }
934296
protected void sendSubscribe(String topic) {
    assert topic != null : "topic shouldn't be null";
    JsonObject msg = Json.createObject().set(TYPE, "register").set(TOPIC, topic);
    send(msg);
  }
904889
private void publishCheckpointEvent(String stepName, long jobInstanceId, long jobExecutionId, long stepExecutionId) {
        BatchEventsPublisher publisher = getBatchEventsPublisher();

        if (publisher != null) {
            String correlationId = runtimeWorkUnitExecution.getCorrelationId();
            publisher.publishCheckpointEvent(stepName, jobInstanceId, jobExecutionId, stepExecutionId, correlationId);
        }
    }
353645
protected void publishJobStarted(CmsPublishJobInfoBean publishJob) {

        // update the job
        m_publishQueue.update(publishJob);

        // fire the publish start event
        m_listeners.fireStart(new CmsPublishJobEnqueued(publishJob));
    }
1195698
def publish(self, topic, *args, **kwargs):
        
        return self._async_session.publish(topic, *args, **kwargs)
700702
private void publishPurgeEvent(WSJobInstance jobInstance) {
        if (eventsPublisher != null) {
            eventsPublisher.publishJobInstanceEvent(jobInstance, BatchEventsPublisher.TOPIC_INSTANCE_PURGED, null);
        }
    }
1212022
private synchronized void updateStatus(JobStatus _newStatus) {

    // generate a new event ID (always increasing).
    JobStatus newStatus = _newStatus.withNewEventId();

    UUID jobId = newStatus.getJobId();

    // remove the last event for the same job, as it is no longer relevant.
    JobStatus oldStatus = statusByJobId.get(jobId);
    if (oldStatus != null) {
      statusByEventId.remove(oldStatus.getEventId());
    }

    // insert event into lookup tables.
    statusByEventId.put(newStatus.getEventId(), newStatus);
    statusByJobId.put(jobId, newStatus);

    // wake up any listeners.
    notifyAll();

  }
275706
@Override
	public CompletableFuture<Acknowledge> submitTask(
			TaskDeploymentDescriptor tdd,
			JobMasterId jobMasterId,
			Time timeout) {

		try {
			final JobID jobId = tdd.getJobId();
			final JobManagerConnection jobManagerConnection = jobManagerTable.get(jobId);

			if (jobManagerConnection == null) {
				final String message = "Could not submit task because there is no JobManager " +
					"associated for the job " + jobId + '.';

				log.debug(message);
				throw new TaskSubmissionException(message);
			}

			if (!Objects.equals(jobManagerConnection.getJobMasterId(), jobMasterId)) {
				final String message = "Rejecting the task submission because the job manager leader id " +
					jobMasterId + " does not match the expected job manager leader id " +
					jobManagerConnection.getJobMasterId() + '.';

				log.debug(message);
				throw new TaskSubmissionException(message);
			}

			if (!taskSlotTable.tryMarkSlotActive(jobId, tdd.getAllocationId())) {
				final String message = "No task slot allocated for job ID " + jobId +
					" and allocation ID " + tdd.getAllocationId() + '.';
				log.debug(message);
				throw new TaskSubmissionException(message);
			}

			// re-integrate offloaded data:
			try {
				tdd.loadBigData(blobCacheService.getPermanentBlobService());
			} catch (IOException | ClassNotFoundException e) {
				throw new TaskSubmissionException("Could not re-integrate offloaded TaskDeploymentDescriptor data.", e);
			}

			// deserialize the pre-serialized information
			final JobInformation jobInformation;
			final TaskInformation taskInformation;
			try {
				jobInformation = tdd.getSerializedJobInformation().deserializeValue(getClass().getClassLoader());
				taskInformation = tdd.getSerializedTaskInformation().deserializeValue(getClass().getClassLoader());
			} catch (IOException | ClassNotFoundException e) {
				throw new TaskSubmissionException("Could not deserialize the job or task information.", e);
			}

			if (!jobId.equals(jobInformation.getJobId())) {
				throw new TaskSubmissionException(
					"Inconsistent job ID information inside TaskDeploymentDescriptor (" +
						tdd.getJobId() + " vs. " + jobInformation.getJobId() + ")");
			}

			TaskMetricGroup taskMetricGroup = taskManagerMetricGroup.addTaskForJob(
				jobInformation.getJobId(),
				jobInformation.getJobName(),
				taskInformation.getJobVertexId(),
				tdd.getExecutionAttemptId(),
				taskInformation.getTaskName(),
				tdd.getSubtaskIndex(),
				tdd.getAttemptNumber());

			InputSplitProvider inputSplitProvider = new RpcInputSplitProvider(
				jobManagerConnection.getJobManagerGateway(),
				taskInformation.getJobVertexId(),
				tdd.getExecutionAttemptId(),
				taskManagerConfiguration.getTimeout());

			TaskManagerActions taskManagerActions = jobManagerConnection.getTaskManagerActions();
			CheckpointResponder checkpointResponder = jobManagerConnection.getCheckpointResponder();
			GlobalAggregateManager aggregateManager = jobManagerConnection.getGlobalAggregateManager();

			LibraryCacheManager libraryCache = jobManagerConnection.getLibraryCacheManager();
			ResultPartitionConsumableNotifier resultPartitionConsumableNotifier = jobManagerConnection.getResultPartitionConsumableNotifier();
			PartitionProducerStateChecker partitionStateChecker = jobManagerConnection.getPartitionStateChecker();

			final TaskLocalStateStore localStateStore = localStateStoresManager.localStateStoreForSubtask(
				jobId,
				tdd.getAllocationId(),
				taskInformation.getJobVertexId(),
				tdd.getSubtaskIndex());

			final JobManagerTaskRestore taskRestore = tdd.getTaskRestore();

			final TaskStateManager taskStateManager = new TaskStateManagerImpl(
				jobId,
				tdd.getExecutionAttemptId(),
				localStateStore,
				taskRestore,
				checkpointResponder);

			Task task = new Task(
				jobInformation,
				taskInformation,
				tdd.getExecutionAttemptId(),
				tdd.getAllocationId(),
				tdd.getSubtaskIndex(),
				tdd.getAttemptNumber(),
				tdd.getProducedPartitions(),
				tdd.getInputGates(),
				tdd.getTargetSlotNumber(),
				taskExecutorServices.getMemoryManager(),
				taskExecutorServices.getIOManager(),
				taskExecutorServices.getNetworkEnvironment(),
				taskExecutorServices.getKvStateService(),
				taskExecutorServices.getBroadcastVariableManager(),
				taskExecutorServices.getTaskEventDispatcher(),
				taskStateManager,
				taskManagerActions,
				inputSplitProvider,
				checkpointResponder,
				aggregateManager,
				blobCacheService,
				libraryCache,
				fileCache,
				taskManagerConfiguration,
				taskMetricGroup,
				resultPartitionConsumableNotifier,
				partitionStateChecker,
				getRpcService().getExecutor());

			log.info("Received task {}.", task.getTaskInfo().getTaskNameWithSubtasks());

			boolean taskAdded;

			try {
				taskAdded = taskSlotTable.addTask(task);
			} catch (SlotNotFoundException | SlotNotActiveException e) {
				throw new TaskSubmissionException("Could not submit task.", e);
			}

			if (taskAdded) {
				task.startTaskThread();

				return CompletableFuture.completedFuture(Acknowledge.get());
			} else {
				final String message = "TaskManager already contains a task for id " +
					task.getExecutionId() + '.';

				log.debug(message);
				throw new TaskSubmissionException(message);
			}
		} catch (TaskSubmissionException e) {
			return FutureUtils.completedExceptionally(e);
		}
	}
1196270
@Override
    public WSJobInstance createJobInstance(String appName, String jobXMLName, String submitter, String jsl, String correlationId) {

        JobInstanceEntity retMe = null;

        retMe = getPersistenceManagerService().createJobInstance(appName,
                                                                 jobXMLName,
                                                                 jsl,
                                                                 submitter,
                                                                 new Date());

        publishEvent(retMe, BatchEventsPublisher.TOPIC_INSTANCE_SUBMITTED, correlationId);

        return retMe;
    }
24525
public static void create(DbConn cnx, JobInstance ji, State finalState, Calendar endDate)
    {
        JobDef jd = ji.getJD();
        Node n = ji.getNode();
        Queue q = ji.getQ();

        if (endDate == null)
        {
            cnx.runUpdate("history_insert", ji.getId(), jd.getApplication(), jd.getApplicationName(), ji.getAttributionDate(),
                    ji.getEmail(), ji.getCreationDate(), ji.getExecutionDate(), jd.isHighlander(), ji.getApplication(), ji.getKeyword1(),
                    ji.getKeyword2(), ji.getKeyword3(), ji.getModule(), jd.getKeyword1(), jd.getKeyword2(), jd.getKeyword3(),
                    jd.getModule(), n == null ? null : n.getName(), ji.getParentId(), ji.getProgress(), q == null ? null : q.getName(), 0,
                    ji.getSessionID(), finalState.toString(), ji.getUserName(), ji.getJdId(), n == null ? null : n.getId(), ji.getQueue(),
                    ji.isFromSchedule(), ji.getPriority());
        }
        else
        {
            cnx.runUpdate("history_insert_with_end_date", ji.getId(), jd.getApplication(), jd.getApplicationName(), ji.getAttributionDate(),
                    ji.getEmail(), endDate, ji.getCreationDate(), ji.getExecutionDate(), jd.isHighlander(), ji.getApplication(),
                    ji.getKeyword1(), ji.getKeyword2(), ji.getKeyword3(), ji.getModule(), jd.getKeyword1(), jd.getKeyword2(),
                    jd.getKeyword3(), jd.getModule(), n.getName(), ji.getParentId(), ji.getProgress(), q.getName(), 0, ji.getSessionID(),
                    finalState.toString(), ji.getUserName(), ji.getJdId(), ji.getNode().getId(), ji.getQueue(), ji.isFromSchedule(),
                    ji.getPriority());
        }
    }
1061136
@Override
    public Topic createTopic(String topicName) throws JMSRuntimeException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createTopic", new Object[] { topicName });

        Topic topic = null;

        try {
            topic = jmsSession.createTopic(topicName);
        } catch (JMSException jmse) {
            throw (JMSRuntimeException) JmsErrorUtils.getJMS2Exception(jmse, JMSRuntimeException.class);
        } finally {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(this, tc, "createTopic", new Object[] { topic });
        }

        return topic;
    }
241348
@Override
	void sendTaskEvent(TaskEvent event) throws IOException {
		checkError();
		checkState(subpartitionView != null, "Tried to send task event to producer before requesting the subpartition.");

		if (!taskEventPublisher.publish(partitionId, event)) {
			throw new IOException("Error while publishing event " + event + " to producer. The producer could not be found.");
		}
	}
821843
void sendMigrationEvent(final MigrationInfo migrationInfo, final MigrationEvent.MigrationStatus status) {
        if (migrationInfo.getSourceCurrentReplicaIndex() != 0
                && migrationInfo.getDestinationNewReplicaIndex() != 0) {
            // only fire events for 0th replica migrations
            return;
        }

        ClusterServiceImpl clusterService = node.getClusterService();
        MemberImpl current = clusterService.getMember(migrationInfo.getSourceAddress());
        MemberImpl newOwner = clusterService.getMember(migrationInfo.getDestinationAddress());
        MigrationEvent event = new MigrationEvent(migrationInfo.getPartitionId(), current, newOwner, status);
        EventService eventService = nodeEngine.getEventService();
        Collection<EventRegistration> registrations = eventService.getRegistrations(SERVICE_NAME, MIGRATION_EVENT_TOPIC);
        eventService.publishEvent(SERVICE_NAME, registrations, event, event.getPartitionId());
    }
501544
public static void processTopicID(final BuildData buildData, final ITopicNode topicNode, final Document doc) {
        final BaseTopicWrapper<?> topic = topicNode.getTopic();

        // Ignore info topics
        if (!topic.hasTag(buildData.getServerEntities().getInfoTagId())) {
            // Get the id value
            final String errorXRefID = ((SpecNode) topicNode).getUniqueLinkId(buildData.isUseFixedUrls());

            // Set the id
            setDOMElementId(buildData.getDocBookVersion(), doc.getDocumentElement(), errorXRefID);
        }

        // Add the remap parameter
        final Integer topicId = topicNode.getDBId();
        doc.getDocumentElement().setAttribute("remap", "TID_" + topicId);
    }