Cannot write msg to topic by kafka console tool when enabled kop protocal

Describe the bug

I got a warn msg when using kafka-console-producer.sh write data to pulsar. ./bin/kafka-console-producer.sh --broker-list A.A.A.A:9092,B.B.B.B:9092,C.C.C.:9092 --topic test_4

[2022-08-05 07:22:56,509] WARN [Producer clientId=console-producer] Got error produce response with correlation id 5 on topic-partition public/default/t_4-0, retrying (2 attempts left). Error: NOT_LEADER_FOR_PARTITION (org.apache.kafka.clients.producer.internals.Sender) [2022-08-05 07:22:56,510] WARN [Producer clientId=console-producer] Received invalid metadata error in produce request on partition public/default/t_4-0 due to org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)

At the same time, pulsar-broker.log got some logs, like

2022-08-05T07:44:42,241+0000 [pulsar-web-36-7] INFO org.eclipse.jetty.server.RequestLog - a.a.a.a - - [05/Aug/2022:07:44:42 +0000] "GET /admin/v2/persistent/public/default/t_4/partitions HTTP/1.1" 200 16 "-" "Pulsar-Java-v2.10.1" 4 2022-08-05T07:44:42,333+0000 [pulsar-ph-kafka-59-3] INFO org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - Opening managed ledger public/default/persistent/t_4-partition-0 2022-08-05T07:44:42,335+0000 [bookkeeper-ml-scheduler-OrderedScheduler-5-0] INFO org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [public/default/persistent/t_4-partition-0] Closing managed ledger 2022-08-05T07:44:42,335+0000 [bookkeeper-ml-scheduler-OrderedScheduler-5-0] WARN org.apache.pulsar.broker.service.BrokerService - Namespace bundle for topic (persistent://public/default/t_4) not served by this instance. Please redo the lookup. Request is denied: namespace=public/default

And Pulsar broker config:

metadataStoreUrl=A.A.A.A:2181,B.B.B.B:2181,C.C.C.C:2181/pulsar_2_10_1 configurationMetadataStoreUrl=A.A.A.A:2181,B.B.B.B:2181,C.C.C.C:2181/pulsar_2_10_1 brokerServicePort=6650 brokerServicePortTls= webServicePort=8080 webServicePortTls= webServiceTlsProtocols= webServiceTlsCiphers= bindAddress=0.0.0.0 bindAddresses= advertisedAddress=A.A.A.A haProxyProtocolEnabled=false numAcceptorThreads= numIOThreads= numOrderedExecutorThreads=8 numHttpServerThreads= numExecutorThreadPoolSize= numCacheExecutorThreadPoolSize=10 enableBusyWait=false maxConcurrentHttpRequests=1024 isRunningStandalone= clusterName=pulsar-cluster-1 maxTenants=0 failureDomainsEnabled=false metadataStoreSessionTimeoutMillis=30000 metadataStoreOperationTimeoutSeconds=30 metadataStoreCacheExpirySeconds=300 brokerShutdownTimeoutMs=60000 skipBrokerShutdownOnOOM=false backlogQuotaCheckEnabled=true backlogQuotaCheckIntervalInSeconds=60 backlogQuotaDefaultLimitBytes=-1 backlogQuotaDefaultLimitSecond=-1 backlogQuotaDefaultRetentionPolicy=producer_request_hold ttlDurationDefaultInSeconds=0 allowAutoTopicCreation=true allowAutoTopicCreationType=non-partitioned allowAutoSubscriptionCreation=true defaultNumPartitions=1 brokerDeleteInactiveTopicsEnabled=false brokerDeleteInactiveTopicsFrequencySeconds=60 brokerDeleteInactiveTopicsMode=delete_when_no_subscriptions brokerDeleteInactivePartitionedTopicMetadataEnabled=false brokerDeleteInactiveTopicsMaxInactiveDurationSeconds= forceDeleteTenantAllowed=false forceDeleteNamespaceAllowed=false maxPendingPublishRequestsPerConnection=1000 messageExpiryCheckIntervalInMinutes=5 activeConsumerFailoverDelayTimeMillis=1000 subscriptionExpirationTimeMinutes=0 subscriptionRedeliveryTrackerEnabled=true subscriptionExpiryCheckIntervalInMinutes=5 subscriptionTypesEnabled=Exclusive,Shared,Failover,Key_Shared subscriptionKeySharedUseConsistentHashing=true subscriptionKeySharedConsistentHashingReplicaPoints=100 brokerDeduplicationEnabled=false brokerDeduplicationMaxNumberOfProducers=10000 brokerDeduplicationSnapshotFrequencyInSeconds=120 brokerDeduplicationSnapshotIntervalSeconds=120 brokerDeduplicationEntriesInterval=1000 brokerDeduplicationProducerInactivityTimeoutMinutes=360 defaultNumberOfNamespaceBundles=4 maxNamespacesPerTenant=0 maxTopicsPerNamespace=0 brokerMaxConnections=0 brokerMaxConnectionsPerIp=0 isAllowAutoUpdateSchemaEnabled=true clientLibraryVersionCheckEnabled=false statusFilePath= preferLaterVersions=false maxUnackedMessagesPerConsumer=50000 maxUnackedMessagesPerSubscription=200000 maxUnackedMessagesPerBroker=0 maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16 unblockStuckSubscriptionEnabled=false topicPublisherThrottlingTickTimeMillis=10 preciseTopicPublishRateLimiterEnable=false brokerPublisherThrottlingTickTimeMillis=50 brokerPublisherThrottlingMaxMessageRate=0 brokerPublisherThrottlingMaxByteRate=0 maxPublishRatePerTopicInMessages=0 maxPublishRatePerTopicInBytes=0 subscribeThrottlingRatePerConsumer=0 subscribeRatePeriodPerConsumerInSecond=30 dispatchThrottlingRateInMsg=0 dispatchThrottlingRateInByte=0 dispatchThrottlingRatePerTopicInMsg=0 dispatchThrottlingRatePerTopicInByte=0 dispatchThrottlingOnBatchMessageEnabled=false dispatchThrottlingRatePerSubscriptionInMsg=0 dispatchThrottlingRatePerSubscriptionInByte=0 dispatchThrottlingRatePerReplicatorInMsg=0 dispatchThrottlingRatePerReplicatorInByte=0 dispatchThrottlingRateRelativeToPublishRate=false dispatchThrottlingOnNonBacklogConsumerEnabled=true dispatcherMaxReadBatchSize=100 dispatcherMaxReadSizeBytes=5242880 dispatcherMinReadBatchSize=1 dispatcherMaxRoundRobinBatchSize=20 dispatcherReadFailureBackoffInitialTimeInMs=15000 dispatcherReadFailureBackoffMaxTimeInMs=60000 dispatcherReadFailureBackoffMandatoryStopTimeInMs=0 preciseDispatcherFlowControl=false entryFilterNames= entryFiltersDirectory= maxConcurrentLookupRequest=50000 maxConcurrentTopicLoadRequest=5000 maxConcurrentNonPersistentMessagePerConnection=1000 numWorkerThreadsForNonPersistentTopic= enablePersistentTopics=true enableNonPersistentTopics=true enableRunBookieTogether=false enableRunBookieAutoRecoveryTogether=false maxProducersPerTopic=0 maxSameAddressProducersPerTopic=0 encryptionRequireOnProducer=false maxConsumersPerTopic=0 maxSameAddressConsumersPerTopic=0 maxSubscriptionsPerTopic=0 maxConsumersPerSubscription=0 maxMessageSize=5242880 brokerServiceCompactionMonitorIntervalInSeconds=60 brokerServiceCompactionThresholdInBytes=0 brokerServiceCompactionPhaseOneLoopTimeInSeconds=30 delayedDeliveryEnabled=true delayedDeliveryTickTimeMillis=1000 acknowledgmentAtBatchIndexLevelEnabled=false enableReplicatedSubscriptions=true replicatedSubscriptionsSnapshotFrequencyMillis=1000 replicatedSubscriptionsSnapshotTimeoutSeconds=30 replicatedSubscriptionsSnapshotMaxCachedPerSubscription=10 maxMessagePublishBufferSizeInMB= retentionCheckIntervalInSeconds=120 maxMessageSizeCheckIntervalInSeconds=60 maxNumPartitionsPerPartitionedTopic=0 zookeeperSessionExpiredPolicy=reconnect systemTopicEnabled=false systemTopicSchemaCompatibilityStrategy=ALWAYS_COMPATIBLE topicLevelPoliciesEnabled=false topicFencingTimeoutSeconds=0 proxyRoles= authenticateOriginalAuthData=false tlsCertRefreshCheckDurationSec=300 tlsCertificateFilePath= tlsKeyFilePath= tlsTrustCertsFilePath= tlsAllowInsecureConnection=false tlsProtocols= tlsCiphers= tlsRequireTrustedClientCertOnConnect=false tlsProvider= webServiceTlsProvider=Conscrypt tlsEnabledWithKeyStore=false tlsKeyStoreType=JKS tlsKeyStore= tlsKeyStorePassword= tlsTrustStoreType=JKS tlsTrustStore= tlsTrustStorePassword= brokerClientTlsEnabledWithKeyStore=false brokerClientSslProvider= brokerClientTlsTrustStoreType=JKS brokerClientTlsTrustStore= brokerClientTlsTrustStorePassword= brokerClientTlsCiphers= brokerClientTlsProtocols= metadataStoreBatchingEnabled=true metadataStoreBatchingMaxDelayMillis=5 metadataStoreBatchingMaxOperations=1000 metadataStoreBatchingMaxSizeKb=128 authenticationEnabled=false authenticationProviders= authenticationRefreshCheckSeconds=60 authorizationEnabled=false authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider authorizationAllowWildcardsMatching=false superUserRoles= brokerClientTlsEnabled=false brokerClientAuthenticationPlugin= brokerClientAuthenticationParameters= brokerClientTrustCertsFilePath= athenzDomainNames= anonymousUserRole= tokenSecretKey= tokenPublicKey= tokenAuthClaim= tokenAudienceClaim= tokenAudience= saslJaasClientAllowedIds= saslJaasServerSectionName= httpMaxRequestSize=-1 disableHttpDebugMethods=false httpRequestsLimitEnabled=false httpRequestsMaxPerSecond=100.0 bookkeeperMetadataServiceUri=zk+null://A.A.A.A:2181,B.B.B.B:2181,C.C.C.C:2181/pulsar_2_10_1/ledgers bookkeeperClientAuthenticationPlugin= bookkeeperClientAuthenticationParametersName= bookkeeperClientAuthenticationParameters= bookkeeperClientTimeoutInSeconds=30 bookkeeperClientNumWorkerThreads= bookkeeperClientSpeculativeReadTimeoutInMillis=0 bookkeeperNumberOfChannelsPerBookie=16 bookkeeperUseV2WireProtocol=true bookkeeperClientHealthCheckEnabled=true bookkeeperClientHealthCheckIntervalSeconds=60 bookkeeperClientHealthCheckErrorThresholdPerInterval=5 bookkeeperClientHealthCheckQuarantineTimeInSeconds=1800 bookkeeperClientQuarantineRatio=1.0 bookkeeperClientGetBookieInfoIntervalSeconds=86400 bookkeeperClientGetBookieInfoRetryIntervalSeconds=60 bookkeeperClientRackawarePolicyEnabled=true bookkeeperClientRegionawarePolicyEnabled=false bookkeeperClientMinNumRacksPerWriteQuorum=2 bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false bookkeeperClientReorderReadSequenceEnabled=false bookkeeperClientIsolationGroups= bookkeeperClientSecondaryIsolationGroups= bookkeeperClientMinAvailableBookiesInIsolationGroups= bookkeeperEnableStickyReads=true bookkeeperTLSProviderFactoryClass=org.apache.bookkeeper.tls.TLSContextFactory bookkeeperTLSClientAuthentication=false bookkeeperTLSKeyFileType=PEM bookkeeperTLSTrustCertTypes=PEM bookkeeperTLSKeyStorePasswordPath= bookkeeperTLSTrustStorePasswordPath= bookkeeperTLSKeyFilePath= bookkeeperTLSCertificateFilePath= bookkeeperTLSTrustCertsFilePath= bookkeeperTlsCertFilesRefreshDurationSeconds=300 bookkeeperDiskWeightBasedPlacementEnabled=false bookkeeperExplicitLacIntervalInMills=0 managedLedgerDefaultEnsembleSize=2 managedLedgerDefaultWriteQuorum=2 managedLedgerDefaultAckQuorum=2 managedLedgerCursorPositionFlushSeconds=60 managedLedgerStatsPeriodSeconds=60 managedLedgerDigestType=CRC32C managedLedgerNumWorkerThreads=8 managedLedgerNumSchedulerThreads=8 managedLedgerCacheSizeMB= managedLedgerCacheCopyEntries=false managedLedgerCacheEvictionWatermark=0.9 managedLedgerCacheEvictionFrequency=100.0 managedLedgerCacheEvictionTimeThresholdMillis=1000 managedLedgerCursorBackloggedThreshold=1000 managedLedgerDefaultMarkDeleteRateLimit=1.0 managedLedgerMaxEntriesPerLedger=50000 managedLedgerMinLedgerRolloverTimeMinutes=10 managedLedgerMaxLedgerRolloverTimeMinutes=240 managedLedgerInactiveLedgerRolloverTimeSeconds=0 managedLedgerMaxSizePerLedgerMbytes=2048 managedLedgerOffloadDeletionLagMs=14400000 managedLedgerOffloadAutoTriggerSizeThresholdBytes=-1 managedLedgerCursorMaxEntriesPerLedger=50000 managedLedgerCursorRolloverTimeInSeconds=14400 managedLedgerMaxUnackedRangesToPersist=10000 managedLedgerMaxUnackedRangesToPersistInZooKeeper=1000 autoSkipNonRecoverableData=false lazyCursorRecovery=false managedLedgerMetadataOperationsTimeoutSeconds=60 managedLedgerReadEntryTimeoutSeconds=0 managedLedgerAddEntryTimeoutSeconds=0 managedLedgerPrometheusStatsLatencyRolloverSeconds=60 managedLedgerTraceTaskExecution=true managedLedgerNewEntriesCheckDelayInMillis=10 loadBalancerEnabled=true loadBalancerReportUpdateThresholdPercentage=10 loadBalancerReportUpdateMaxIntervalMinutes=15 loadBalancerHostUsageCheckIntervalMinutes=1 loadBalancerSheddingEnabled=true loadBalancerSheddingIntervalMinutes=1 loadBalancerSheddingGracePeriodMinutes=30 loadBalancerBrokerMaxTopics=50000 loadBalancerBrokerOverloadedThresholdPercentage=85 loadBalancerResourceQuotaUpdateIntervalMinutes=15 loadBalancerAutoBundleSplitEnabled=true loadBalancerAutoUnloadSplitBundlesEnabled=true loadBalancerNamespaceBundleMaxTopics=1000 loadBalancerNamespaceBundleMaxSessions=1000 loadBalancerNamespaceBundleMaxMsgRate=30000 loadBalancerNamespaceBundleMaxBandwidthMbytes=100 loadBalancerNamespaceMaximumBundles=128 loadBalancerOverrideBrokerNicSpeedGbps=10 loadManagerClassName=org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl supportedNamespaceBundleSplitAlgorithms=range_equally_divide,topic_count_equally_divide defaultNamespaceBundleSplitAlgorithm=range_equally_divide loadBalancerLoadSheddingStrategy=org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder loadBalancerBrokerThresholdShedderPercentage=10 loadBalancerMsgRateDifferenceShedderThreshold=50 loadBalancerMsgThroughputMultiplierDifferenceShedderThreshold=4 loadBalancerHistoryResourcePercentage=0.9 loadBalancerBandwithInResourceWeight=1.0 loadBalancerBandwithOutResourceWeight=1.0 loadBalancerCPUResourceWeight=1.0 loadBalancerMemoryResourceWeight=1.0 loadBalancerDirectMemoryResourceWeight=1.0 loadBalancerBundleUnloadMinThroughputThreshold=10 namespaceBundleUnloadingTimeoutMs=60000 replicationMetricsEnabled=true replicationConnectionsPerBroker=16 replicationProducerQueueSize=1000 replicatorPrefix=pulsar.repl replicationPolicyCheckDurationSeconds=600 defaultRetentionTimeInMinutes=0 defaultRetentionSizeInMB=0 keepAliveIntervalSeconds=30 bootstrapNamespaces= webSocketServiceEnabled=false webSocketNumIoThreads=8 webSocketConnectionsPerBroker=8 webSocketSessionIdleTimeoutMillis=300000 webSocketMaxTextFrameSize=1048576 exposeTopicLevelMetricsInPrometheus=true exposeConsumerLevelMetricsInPrometheus=false exposeProducerLevelMetricsInPrometheus=false exposeManagedLedgerMetricsInPrometheus=true exposeManagedCursorMetricsInPrometheus=false metricsServletTimeoutMs=30000 exposeBundlesMetricsInPrometheus=false functionsWorkerEnabled=true exposePublisherStats=true statsUpdateFrequencyInSecs=60 statsUpdateInitialDelayInSecs=60 exposePreciseBacklogInPrometheus=false splitTopicAndPartitionLabelInPrometheus=false aggregatePublisherStatsByProducerName=false schemaRegistryStorageClassName=org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorageFactory isSchemaValidationEnforced=false schemaCompatibilityStrategy=FULL offloadersDirectory=./offloaders managedLedgerOffloadDriver= managedLedgerOffloadMaxThreads=2 managedLedgerOffloadPrefetchRounds=1 managedLedgerUnackedRangesOpenCacheSetEnabled=true s3ManagedLedgerOffloadRegion= s3ManagedLedgerOffloadBucket= s3ManagedLedgerOffloadServiceEndpoint= s3ManagedLedgerOffloadMaxBlockSizeInBytes=67108864 s3ManagedLedgerOffloadReadBufferSizeInBytes=1048576 gcsManagedLedgerOffloadRegion= gcsManagedLedgerOffloadBucket= gcsManagedLedgerOffloadMaxBlockSizeInBytes=67108864 gcsManagedLedgerOffloadReadBufferSizeInBytes=1048576 gcsManagedLedgerOffloadServiceAccountKeyFile= managedLedgerOffloadBucket= fileSystemProfilePath=../conf/filesystem_offload_core_site.xml fileSystemURI= transactionCoordinatorEnabled=false transactionMetadataStoreProviderClassName=org.apache.pulsar.transaction.coordinator.impl.MLTransactionMetadataStoreProvider transactionBufferSnapshotMaxTransactionCount=1000 transactionBufferSnapshotMinTimeInMillis=5000 transactionBufferClientMaxConcurrentRequests=1000 transactionPendingAckLogIndexMinLag=500 enablePackagesManagement=false packagesManagementStorageProvider=org.apache.pulsar.packages.management.storage.bookkeeper.BookKeeperPackagesStorageProvider packagesReplicas=1 packagesManagementLedgerRootPath=/ledgers zookeeperServers= globalZookeeperServers= configurationStoreServers= zooKeeperSessionTimeoutMillis=-1 zooKeeperOperationTimeoutSeconds=-1 zooKeeperCacheExpirySeconds=-1 replicationTlsEnabled=false brokerServicePurgeInactiveFrequencyInSeconds=60 backlogQuotaDefaultLimitGB=-1 tlsEnabled=false subscriptionKeySharedEnable=true messagingProtocols=kafka protocolHandlerDirectory=./protocols allowAutoTopicCreationType=partitioned kafkaListeners=PLAINTEXT://0.0.0.0:9092 kafkaAdvertisedListeners=PLAINTEXT://A.A.A.A:9092 brokerEntryMetadataInterceptors=org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor

Env (please complete the following information):

pulsar: 2.10.1 , 3 nodes cluster(broker + bookie + zk)
kop: pulsar-protocol-handler-kafka-2.10.1.4.nar
kafka: 2.0.0

apache / pulsar

Cannot write msg to topic by kafka console tool when enabled kop protocal #16956