ClickHouse 无法从 Kafka 获取 TabSeparated 格式的消息。

huangapple go评论74阅读模式
英文:

Clickhouse can't get messages from Kafka with TabSeparated formatting

问题

I send messages to Kafka from my Spring Boot application:

ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("uniqTopic123", "testKey", "Test\tTest");
future.addCallback(
        (v) -> System.out.println("SUCCESS: " + v),
        (v) -> System.out.println("FAIL: " + v)
);
kafkaTemplate.flush();

application.properties:

spring.kafka.consumer.group-id=app.1
kafka.server=<kafka_host>:9092
kafka.producer.id=kafkaProducerId

Configuration:

@Value("${kafka.server}")
private String kafkaServer;

@Value("${kafka.producer.id}")
private String kafkaProducerId;

@Value("${spring.kafka.consumer.group-id}")
private String kafkaGroupId;

@Bean
public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    //props.put(ProducerConfig.CLIENT_ID_CONFIG, kafkaProducerId);
    return props;
}

@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
    return new KafkaTemplate<String, String>(
            new DefaultKafkaProducerFactory<>(producerConfigs()));
}

@Bean
public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaGroupId);

    return new DefaultKafkaConsumerFactory<>(props);
}

In logs I can see messages like:

SUCCESS: SendResult [producerRecord=ProducerRecord(topic=uniqTopic123, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=testKey, value=Test	Test, timestamp=null), recordMetadata=uniqTopic123-0@1]

But my listener doesn't catch any messages:

@KafkaListener(topics="uniqTopic123")
public void msgListener(ConsumerRecord<String, String> record){
    System.out.println("test ======> " + record.value());
}

And the table in ClickHouse is empty. My ClickHouse tables:

CREATE TABLE IF NOT EXISTS test (key String, message String)
ENGINE = Kafka('<kafka_host>:9092', 'uniqTopic123', 'app.1', 'TabSeparated');

CREATE TABLE IF NOT EXISTS test_table (key String, message String)
ENGINE = MergeTree() ORDER BY key;

CREATE MATERIALIZED VIEW consumer TO test_table AS
SELECT key, message FROM test;

What is wrong in my code?

UPD.: Kafka Tool shows that messages are in Kafka

UPD: The mistake was in the absence of a linefeed at the end of the message "Test\tTest\n".

英文:

I send messages to Kafka from my Spring Boot application

ListenableFuture&lt;SendResult&lt;String, String&gt;&gt; future = kafkaTemplate.send(&quot;uniqTopic123&quot;, &quot;testKey&quot;, &quot;Test\tTest&quot;);
future.addCallback(
(v) -&gt; System.out.println(&quot;SUCCESS: &quot; + v),
(v) -&gt; System.out.println(&quot;FAIL: &quot; + v)
);
kafkaTemplate.flush();

application.properties

spring.kafka.consumer.group-id=app.1
kafka.server=&lt;kafka_host&gt;:9092
kafka.producer.id=kafkaProducerId

Configuration

@Value(&quot;${kafka.server}&quot;)
private String kafkaServer;
@Value(&quot;${kafka.producer.id}&quot;)
private String kafkaProducerId;
@Value(&quot;${spring.kafka.consumer.group-id}&quot;)
private String kafkaGroupId;
@Bean
public Map&lt;String, Object&gt; producerConfigs() {
Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//props.put(ProducerConfig.CLIENT_ID_CONFIG, kafkaProducerId);
return props;
}
@Bean
public KafkaTemplate&lt;String, String&gt; kafkaTemplate() {
return new KafkaTemplate&lt;String, String&gt;(
new DefaultKafkaProducerFactory&lt;&gt;(producerConfigs()));
}
@Bean
public KafkaListenerContainerFactory&lt;?&gt; kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
factory.setConsumerFactory(consumerFactory());
return factory;
}
@Bean
public ConsumerFactory&lt;String, String&gt; consumerFactory() {
Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaGroupId);
return new DefaultKafkaConsumerFactory&lt;&gt;(props);
}

In logs I can see messages like:

SUCCESS: SendResult [producerRecord=ProducerRecord(topic=uniqTopic123, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=testKey, value=Test Test, timestamp=null), recordMetadata=uniqTopic123-0@1]

But my listener doesn't catch any messages

@KafkaListener(topics=&quot;uniqTopic123&quot;)
public void msgListener(ConsumerRecord&lt;String, String&gt; record){
System.out.println(&quot;test ======&gt; &quot; + record.value());
}

And the table in ClickHouse is empty. My ClickHouse tables

CREATE TABLE IF NOT EXISTS test (key String, message String)
ENGINE = Kafka(&#39;&lt;kafka_host&gt;:9092&#39;, &#39;uniqTopic123&#39;, &#39;app.1&#39;, &#39;TabSeparated&#39;);
CREATE TABLE IF NOT EXISTS test_table (key String, message String)
ENGINE = MergeTree() ORDER BY key;
CREATE MATERIALIZED VIEW consumer TO test_table AS
SELECT key, message FROM test;

What is wrong in my code?

UPD.: Kafka Tool shows that messages are in Kafka
ClickHouse 无法从 Kafka 获取 TabSeparated 格式的消息。

UPD:
The mistake was in absenсe of linefeed at the end of message "Test\tTest\n"

答案1

得分: 1

错误在于消息末尾没有换行符,消息为"Test\tTest\n"。

英文:

The mistake was in absenсe of linefeed at the end of message "Test\tTest\n"

huangapple
  • 本文由 发表于 2020年8月7日 19:27:05
  • 转载请务必保留本文链接:https://go.coder-hub.com/63300872.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定