kafka 생산자/소비자가 다시 시작된 후 소비자가 메시지를 수신하지 않음

하나의 생산자 & 한 명의 생산자가 있습니다. & 한 파티션. 소비자/생산자 모두 스프링 부트 응용 프로그램입니다. 소비자 응용 프로그램은 내 로컬 컴퓨터에서 실행되며 생산자는 kafka & 사육사와 함께 원격 컴퓨터에서 실행됩니다.kafka 생산자/소비자가 다시 시작된 후 소비자가 메시지를 수신하지 않음

개발하는 동안 일부 제작자 응용 프로그램을 재배포했습니다. 하지만 그 후 내 소비자는 어떤 메시지도받지 못합니다. 나는 소비자 재출발을 시도했지만 운이 없다. 문제는 무엇이 될 수 있으며 어떻게 해결 될 수 있습니까?

소비자 구성 :

spring: 
    cloud: 
    stream: 
     defaultBinder: kafka 
     bindings: 
     input: 
      destination: sales 
      content-type: application/json 
     kafka: 
     binder: 
      brokers: ${SERVICE_REGISTRY_HOST:127.0.0.1} 
      zkNodes: ${SERVICE_REGISTRY_HOST:127.0.0.1} 
      defaultZkPort: 2181 
      defaultBrokerPort: 9092 
server: 
    port: 0

프로듀서 구성 :

cloud: 
stream: 
    defaultBinder: kafka 
    bindings: 
    output: 
     destination: sales 
     content-type: application/json 
    kafka: 
    binder: 
     brokers: ${SERVICE_REGISTRY_HOST:127.0.0.1} 
     zkNodes: ${SERVICE_REGISTRY_HOST:127.0.0.1} 
     defaultZkPort: 2181 
     defaultBrokerPort: 9092

EDIT2 : 소비자 응용 프로그램은 다음과 같은 예외로 사망

5 분 후 :

2017-09-12 18:14:47,254 ERROR main o.s.c.s.b.k.p.KafkaTopicProvisioner:253 - Cannot initialize Binder 
org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata 
2017-09-12 18:14:47,255 WARN main o.s.b.c.e.AnnotationConfigEmbeddedWebApplicationContext:550 - Exception encountered during context initialization - cancelling refresh attempt: org.springframework.context.ApplicationContextException: Failed to start bean 'inputBindingLifecycle'; nested exception is org.springframework.cloud.stream.binder.BinderException: Cannot initialize binder: 
2017-09-12 18:14:47,256 INFO main o.s.i.m.IntegrationMBeanExporter:449 - Unregistering JMX-exposed beans on shutdown 
2017-09-12 18:14:47,257 INFO main o.s.i.m.IntegrationMBeanExporter:241 - Unregistering JMX-exposed beans 
2017-09-12 18:14:47,257 INFO main o.s.i.m.IntegrationMBeanExporter:375 - Summary on shutdown: input 
2017-09-12 18:14:47,257 INFO main o.s.i.m.IntegrationMBeanExporter:375 - Summary on shutdown: nullChannel 
2017-09-12 18:14:47,258 INFO main o.s.i.m.IntegrationMBeanExporter:375 - Summary on shutdown: errorChannel

출처

2017-09-12 LazyTechie

아주 간단한 시나리오처럼 들립니다. 문제를 로컬에서 재현 할 수 있도록 GitHub 어딘가에서 그 응용 프로그램을 공유 하시겠습니까? –

@ArtemBilan 죄송하지만 코드를 공유 할 수 없습니다. 솔루션을 제안하기 위해 필요한 세부 사항은 무엇입니까? – LazyTechie

코드없이 아이디어가 없습니다. 어쩌면 소비자와 생산자를위한 설정을 공유 할 수 있을까요? 그리고 예, 당신은 전체 애플리케이션을 공유 할 수 없다는 것을 알고 있습니다 만, 적어도, 우리에게 간단한 Spring Boot app를 제공 할 수는 있습니다 ... –

글쎄, resetOffset 속성이 아무 효과가 없다는 것을 나타내는 spring-cloud-stream-binder-kafka으로보고 된 bug이 이미있는 것처럼 보입니다. 따라서 소비자는 항상 오프셋이 latest 인 메시지를 요청했습니다.

git 문제에서 언급했듯이 유일한 해결 방법은 kafka 소비자 CLI 도구를 통해이를 해결하는 것입니다.

출처

2017-09-14 11:49:17 LazyTechie

DEBUG에 관한 위의 제안 사항이 더 자세한 정보를 제공하는지 확인하십시오. KafkaTopicProvisioner에서 일부 시간 초과 예외가 발생하는 것 같습니다. 그러나 그것은 제가 생각하기에 소비자를 다시 시작할 때 발생합니다. 소비자가 어떻게 든 브로커와 의사 소통하는 데 어려움을 겪고 있으며 어떻게 진행되고 있는지 찾아야합니다.

출처

2017-09-12 13:51:58 sobychacko

kafka 생산자/소비자가 다시 시작된 후 소비자가 메시지를 수신하지 않음

답변

관련 문제