apache / apache/pulsar

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

Open
#16,956 5 comments 0 reactions 0 assignees View on GitHub
Stale type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

**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

Contributor guide

Open the contributing guide

Research direction

Start by reproducing the failure with bin/kafka-console-producer.sh and the broker addresses and topic shown in the report. Compare the producer's NOT_LEADER_FOR_PARTITION warnings with the Pulsar broker log entries for the Kafka protocol request and namespace-bundle lookup; done means the console tool can publish successfully without the repeated metadata error.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
cli, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
20/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.