Unable to start queue processor. When i check logs its showing KAFKA connection issue.

<pagedata>
<pzAttempts>0</pzAttempts>
<pzRunInSystemRuntimeContext>false</pzRunInSystemRuntimeContext>
<pzMessageID>f5a48b8a-6f7b-4ec7-8460-76245e162c58</pzMessageID>
<pzMessageClass>System-Queue-FTSBatchReindexMessage</pzMessageClass>
<pxCreateDateTime>20221006T220238.726 GMT</pxCreateDateTime>
<pzAccessGroup>ESI-Base:Agents</pzAccessGroup>
<pyItemStatus>SCHEDULED</pyItemStatus>
<pzEnqueueTimestamp>1665093758726</pzEnqueueTimestamp>
<pxInsName>F5A48B8A-6F7B-4EC7-8460-76245E162C58</pxInsName>
<pxSaveDateTime>20221006T220248.732 GMT</pxSaveDateTime>
<pzAcquireLockOn>NONE</pzAcquireLockOn>
<pzInsKey>SYSTEM-MESSAGE-QUEUEPROCESSOR-DELAYEDITEM F5A48B8A-6F7B-4EC7-8460-76245E162C58</pzInsKey>
<pzAcquireLockTimeout>0</pzAcquireLockTimeout>
<pzQueueProcessorName>pzStandardProcessor</pzQueueProcessorName>
<pzAcquireLockAttempts>0</pzAcquireLockAttempts>
<pxObjClass>System-Message-QueueProcessor-DelayedItem</pxObjClass>
<pzActivityToRun>pzFTSBatchIndexMessenger</pzActivityToRun>
<pzTenantId>shared</pzTenantId>
<pzTimeForProcessing>20221006T220248.729 GMT</pzTimeForProcessing>
<pxCommitDateTime>20221007T020248.000 GMT</pxCommitDateTime>
<pxQueuingNodeId>pega-stream-79bb9557b9-tz6l7</pxQueuingNodeId>
<pzEnqueuedSnapShotOfPage>true</pzEnqueuedSnapShotOfPage>
<pzStatus>valid</pzStatus>
<pyAlertsConfiguration REPEATINGTYPE="PageGroup">
<rowdata REPEATINGINDEX="PEGA0117">
<pyID>PEGA0117</pyID>
<pyIsEnabled>true</pyIsEnabled>
<pyThreshold>2147483647</pyThreshold>
<pxSubscript>PEGA0117</pxSubscript>
</rowdata>
</pyAlertsConfiguration>
<pzProcessingErrorDetails REPEATINGTYPE="PageList">
<rowdata REPEATINGINDEX="1">
<pxObjClass>Embed-QueueErrorDetails</pxObjClass>
<pxNodeId>pega-stream-79bb9557b9-tz6l7</pxNodeId>
<pxExceptionCause>java.util.concurrent.TimeoutException
at java.base/java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1886)
at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2021)
at com.pega.dsm.dnode.impl.dataset.stream.StreamServicePublisherFeature.getPublisher(StreamServicePublisherFeature.java:49)
at com.pega.dsm.dnode.impl.dataset.stream.StreamSaveOperation.save(StreamSaveOperation.java:109)
at com.pega.dsm.dnode.impl.dataset.stream.StreamSaveOperation.doSave(StreamSaveOperation.java:99)
at com.pega.dsm.dnode.impl.dataset.stream.StreamSaveOperation.lambda$null$0(StreamSaveOperation.java:92)
at com.pega.dsm.dnode.impl.dataset.kafka.metrics.KafkaSaveMetrics.timeSave(KafkaSaveMetrics.java:85)
at com.pega.dsm.dnode.impl.dataset.stream.StreamSaveOperation.lambda$save$1(StreamSaveOperation.java:92)
at com.pega.dsm.dnode.impl.stream.DataObservableImpl$SafeDataSubscriber.subscribe(DataObservableImpl.java:353)
at com.pega.dsm.dnode.impl.stream.DataObservableImpl.subscribe(DataObservableImpl.java:55)
at com.pega.dsm.dnode.impl.dataflow.asyncexecutor.QueueProcessorDataSetOperationsImpl.saveToDataSet(QueueProcessorDataSetOperationsImpl.java:208)
at com.pega.dsm.dnode.impl.dataflow.asyncexecutor.QueueProcessorDataSetOperationsImpl.queueToDataSet(QueueProcessorDataSetOperationsImpl.java:170)
at com.pega.dsm.dnode.impl.dataflow.asyncexecutor.QueueProcessorDataSetOperationsImpl.queueToDataSet(QueueProcessorDataSetOperationsImpl.java:150)
at com.pega.platform.executor.queueprocessor.internal.StandardQueueProcessorImpl.enqueue(StandardQueueProcessorImpl.java:58)
at com.pega.platform.executor.queueprocessor.internal.AbstractQueueProcessor.enqueue(AbstractQueueProcessor.java:109)
at com.pega.platform.executor.queueprocessor.internal.StandardQueueProcessorImpl.enqueue(StandardQueueProcessorImpl.java:44)
at com.pega.pegarules.priv.search.nextgen.BatchIndexingQPQueue.enqueueStandard(BatchIndexingQPQueue.java:56)
at com.pega.pegarules.priv.search.nextgen.BatchIndexingQPQueue.enqueueBaseReIndexMessage(BatchIndexingQPQueue.java:26)
at com.pega.platform.pegasearch.indexer.internal.PegaIndexerImpl.enqueueReindexMessage(PegaIndexerImpl.java:631)
at com.pega.platform.pegasearch.indexer.internal.PegaIndexerImpl.buildIndex(PegaIndexerImpl.java:239)
at com.pega.platform.pegasearch.indexer.internal.PegaIndexerImpl.buildIndex(PegaIndexerImpl.java:207)
at com.pega.platform.pegasearch.indexer.internal.PegaIndexerImpl.buildIndex(PegaIndexerImpl.java:202)
at com.pega.platform.pegasearch.searchmanager.internal.BuildIndicesTask$ProcessBuildIndices.processOOTBIndexes(BuildIndicesTask.java:152)
at com.pega.platform.pegasearch.searchmanager.internal.BuildIndicesTask$ProcessBuildIndices.triggerIndexing(BuildIndicesTask.java:120)
at com.pega.platform.pegasearch.searchmanager.internal.BuildIndicesTask$ProcessBuildIndices.run(BuildIndicesTask.java:101)
at java.base/java.util.TimerThread.mainLoop(Timer.java:556)
at java.base/java.util.TimerThread.run(Timer.java:506)
</pxExceptionCause>
<pxException>com.pega.dsm.dnode.impl.dataflow.asyncexecutor.QueueProcessorDataSubscriberException: [QueueProcessor: pzStandardProcessor] [MessageID: f5a48b8a-6f7b-4ec7-8460-76245e162c58] Failed to save message in kafka. Original exception message: null
at com.pega.dsm.dnode.impl.dataflow.asyncexecutor.QueueProcessorDataSubscriber.lambda$onNext$0(QueueProcessorDataSubscriber.java:57)
at java.base/java.util.Optional.map(Optional.java:265)
at com.pega.dsm.dnode.impl.dataflow.asyncexecutor.QueueProcessorDataSubscriber.onNext(QueueProcessorDataSubscriber.java:57)
at com.pega.dsm.dnode.impl.dataflow.asyncexecutor.QueueProcessorDataSubscriber.onNext(QueueProcessorDataSubscriber.java:32)
at com.pega.dsm.dnode.impl.stream.DataObservableImpl$SafeDataSubscriber.onNext(DataObservableImpl.java:305)
at com.pega.dsm.dnode.impl.dataset.stream.StreamSaveOperation.save(StreamSaveOperation.java:115)
at com.pega.dsm.dnode.impl.dataset.stream.StreamSaveOperation.doSave(StreamSaveOperation.java:99)
at com.pega.dsm.dnode.impl.dataset.stream.StreamSaveOperation.lambda$null$0(StreamSaveOperation.java:92)
at com.pega.dsm.dnode.impl.dataset.kafka.metrics.KafkaSaveMetrics.timeSave(KafkaSaveMetrics.java:85)
at com.pega.dsm.dnode.impl.dataset.stream.StreamSaveOperation.lambda$save$1(StreamSaveOperation.java:92)
at com.pega.dsm.dnode.impl.stream.DataObservableImpl$SafeDataSubscriber.subscribe(DataObservableImpl.java:353)
at com.pega.dsm.dnode.impl.stream.DataObservableImpl.subscribe(DataObservableImpl.java:55)
at com.pega.dsm.dnode.impl.dataflow.asyncexecutor.QueueProcessorDataSetOperationsImpl.saveToDataSet(QueueProcessorDataSetOperationsImpl.java:208)
at com.pega.dsm.dnode.impl.dataflow.asyncexecutor.QueueProcessorDataSetOperationsImpl.queueToDataSet(QueueProcessorDataSetOperationsImpl.java:170)
at com.pega.dsm.dnode.impl.dataflow.asyncexecutor.QueueProcessorDataSetOperationsImpl.queueToDataSet(QueueProcessorDataSetOperationsImpl.java:150)
at com.pega.platform.executor.queueprocessor.internal.StandardQueueProcessorImpl.enqueue(StandardQueueProcessorImpl.java:58)
at com.pega.platform.executor.queueprocessor.internal.AbstractQueueProcessor.enqueue(AbstractQueueProcessor.java:109)
at com.pega.platform.executor.queueprocessor.internal.StandardQueueProcessorImpl.enqueue(StandardQueueProcessorImpl.java:44)
at com.pega.pegarules.priv.search.nextgen.BatchIndexingQPQueue.enqueueStandard(BatchIndexingQPQueue.java:56)
at com.pega.pegarules.priv.search.nextgen.BatchIndexingQPQueue.enqueueBaseReIndexMessage(BatchIndexingQPQueue.java:26)
at com.pega.platform.pegasearch.indexer.internal.PegaIndexerImpl.enqueueReindexMessage(PegaIndexerImpl.java:631)
at com.pega.platform.pegasearch.indexer.internal.PegaIndexerImpl.buildIndex(PegaIndexerImpl.java:239)
at com.pega.platform.pegasearch.indexer.internal.PegaIndexerImpl.buildIndex(PegaIndexerImpl.java:207)
at com.pega.platform.pegasearch.indexer.internal.PegaIndexerImpl.buildIndex(PegaIndexerImpl.java:202)
at com.pega.platform.pegasearch.searchmanager.internal.BuildIndicesTask$ProcessBuildIndices.processOOTBIndexes(BuildIndicesTask.java:152)
at com.pega.platform.pegasearch.searchmanager.internal.BuildIndicesTask$ProcessBuildIndices.triggerIndexing(BuildIndicesTask.java:120)
at com.pega.platform.pegasearch.searchmanager.internal.BuildIndicesTask$ProcessBuildIndices.run(BuildIndicesTask.java:101)
at java.base/java.util.TimerThread.mainLoop(Timer.java:556)
at java.base/java.util.TimerThread.run(Timer.java:506)
Caused by: java.util.concurrent.TimeoutException
at java.base/java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1886)
at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2021)
at com.pega.dsm.dnode.impl.dataset.stream.StreamServicePublisherFeature.getPublisher(StreamServicePublisherFeature.java:49)
at com.pega.dsm.dnode.impl.dataset.stream.StreamSaveOperation.save(StreamSaveOperation.java:109)
... 23 more
</pxException>
<pxErrorDateTime>20221006T220248.728 GMT</pxErrorDateTime>
<pxExceptionMessage>[QueueProcessor: pzStandardProcessor] [MessageID: f5a48b8a-6f7b-4ec7-8460-76245e162c58] Failed to save message in kafka. Original exception message: null</pxExceptionMessage>
<pxAuthDetails>
<pxCurrentRulesetList>[ESI-Base:01-23-61|Communication:01-23|CommunicationInt:01-23|Financial:01-23|FinancialInt:01-23|LegacyFinancialInt:01-23|Claim:01-23-61|ClaimInt:01-23-61|ClaimsPrediciton:01-23-63|ClaimsPredicitonInt:01-23-62|Payer:01-23-62|PayerInt:01-23-61|Referral:01-23|ReferralInt:01-23|DocRules:01-23|Ingestion:01-23-61|IngestionInt:01-23-61|DocView:01-23-61|DocViewInt:01-23|Prescription:01-23-73|PrescriptionInt:01-23-62|OrderInt:01-23-64|Outreach:01-23|Drug:01-23|DrugInt:01-23|Ancillary:01-23|AncillaryInt:01-23|Therapy:01-23|TherapyInt:01-23|DrugManufacturer:01-23|DrugManufacturerInt:01-23|Provider:01-23-61|ProviderInt:01-23|Patient:01-23-78|PatientInt:01-23-62|LegacyPatientInt:01-23|Person:01-23-61|ESIHC-Base:01-23|ESIHC-BaseInt:01-23|HCGeneral:01-23|HCGeneralInt:01-23|Protocols:01-23-67|ProtocolsInt:01-23-61|ServiceBranch:01-23|ServiceBranchInt:01-23|ANI:01-23|ANiInt:01-23|Task:01-23|TaskInt:01-23|ESICommon:01-23-61|ESICommonInt:01-23|GenericLookup:01-23|NextActionFW:01-23|Events:01-23-62|EventInt:01-23-61|ServiceNow:01-23-61|ServiceNowInt:01-23|ESISecurity:01-23|ESIOverrides:01-23|ESI:01-23-66|ESIInt:01-23-66|UI-Kit-7:15-01-01|OAuthOverride:01-23|USPSInt:01-23|CAQHProviderInt:01-23|COE-Utilities:01-23|COE-Base:01-23|COE-Monitoring:01-23|Pega-Developer:01-23|PegaHC-HealthCheck:08-06|PegaHC-Ext:08-06|PegaHCFhir-HealthCheck:08-06|PegaHC-FHIR:08-06|Fhir:08-06|PegaX125010-HealthCheck:08-06|PegaEnrollment:08-06|PegaNCPDP:08-06|PegaHCEDI:08-06|PegaX125010:08-06|PegaX12:08-06|PegaX125010-FV:08-06|PegaHCAGM-HealthCheck:08-06|PegaHCAGM:08-06|PegaHCEWS-HealthCheck:08-06|PegaHC-Exchange:08-06|PegaHC-EWS:08-06|PegaHL7-HealthCheck:08-06|PegaHL7:08-06|PegaHCDDS-HealthCheck:08-06|PegaHCDDSInt:08-06|PegaHC-DDS:08-06|PegaHCCore-HealthCheck:08-06|PegaHCMember:08-06|PegaHC-API:08-06|PegaHCLP:08-06|PegaHC:08-06|PegaHC-FV:08-06|PegaHC-Baseclass:08-06|PegaHealthCodeGroups:08-06|PegaHealthCodes:08-06|HCIF-UI:08-06|PegaHealthCodes-FV:08-06|Pega-ProcessCommander:08-06|Pega-DeploymentDefaults:08-06|Pega-DecisionSimulation:08-06|Pega-DecisionScience:08-06|Pega-DecisionArchitect:08-06|Pega-LP-Mobile:08-06|Pega-LP-ProcessAndRules:08-06|Pega-LP-Integration:08-06|Pega-LP-Reports:08-06|Pega-LP-SystemSettings:08-06|Pega-LP-UserInterface:08-06|Pega-LP-OrgAndSecurity:08-06|Pega-LP-DataModel:08-06|Pega-LP-Application:08-06|Pega-LP:08-06|Pega-SystemOperations:08-06|Pega-UpdateManager:08-06|Pega-SecurityVA:08-06|Pega-Feedback:08-06|Pega-HealthCheck:08-06|Pega-AutoTest:08-06|Pega-AppDefinition:08-06|Pega-ImportExport:08-06|Pega-LocalizationTools:08-06|Pega-RuleRefactoring:08-06|Pega-ProcessArchitect:08-06|Pega-Portlet:08-06|Pega-Content:08-06|Pega-BigData:08-06|Pega-NLP:08-06|Pega-DecisionEngine:08-06|Pega-IntegrationArchitect:08-06|Pega-SystemArchitect:08-06|Pega-Desktop:08-06|Pega-EndUserUI:08-06|Pega-Survey:08-06|Pega-Social:08-06|Pega-SharedData:08-06|Pega-API:08-06|Pega-EventProcessing:08-06|Pega-Reporting:08-06|Pega-UIDesign:08-06|Pega-Gadgets:08-06|Pega-UIComponents:08-06|Pega-UIEngine:08-06|Pega-ProcessEngine:08-06|Pega-SearchEngine:08-06|Pega-IntegrationEngine:08-06|Pega-RulesEngine:08-06|Pega-Engine:08-06|PegaLegacyRules:01-01|Pega-ProCom:08-06|Pega-IntSvcs:08-06|Pega-WB:08-06|Pega-RULES:08-06|]</pxCurrentRulesetList>
<pxCurrentRoleList>[PegaRULES:Batch]</pxCurrentRoleList>
<pxCurrentAccessGroup>ESI-Base:Agents</pxCurrentAccessGroup>
</pxAuthDetails>
</rowdata>
</pzProcessingErrorDetails>
<pzMessageContent>
<pxObjClass>System-Queue-FTSBatchReindexMessage</pxObjClass>
<pxWorkClass>System-Queue-FTSBatchReindexMessage</pxWorkClass>
<indexType>Work</indexType>
<pyIndexName>work</pyIndexName>
<pyIndexDedicated>false</pyIndexDedicated>
<pyIndexMode>ALL</pyIndexMode>
<pyFilteredClassList REPEATINGTYPE="PropertyList"/>
</pzMessageContent>
<pzBasicOperatorInfo>
</pzBasicOperatorInfo>
</pagedata>