Hello Matthias and others
I am trying to configure a Kafka Sink with SSL properties as shown further
below.
But in the logs I see warnings:
2022-03-21 12:30:17,108 WARN
org.apache.kafka.clients.admin.AdminClientConfig [] - The
configuration 'group.id' was supplied but isn't a known config.
2022-03-21 12:30:17,109 WARN
org.apache.kafka.clients.admin.AdminClientConfig [] - The
configuration 'partition.discovery.interval.ms' was supplied but isn't a
known config.
2022-03-21 12:30:17,109 WARN
org.apache.kafka.clients.admin.AdminClientConfig [] - The
configuration 'auto.commit.interval.ms' was supplied but isn't a known
config.
2022-03-21 12:30:17,109 WARN
org.apache.kafka.clients.admin.AdminClientConfig [] - The
configuration 'ssl.truststore.type' was supplied but isn't a known config.
2022-03-21 12:30:17,111 WARN
org.apache.kafka.clients.admin.AdminClientConfig [] - The
configuration 'ssl.truststore.location' was supplied but isn't a known
config.
2022-03-21 12:30:17,115 WARN
org.apache.kafka.clients.admin.AdminClientConfig [] - The
configuration 'ssl.truststore.password' was supplied but isn't a known
config.
2022-03-21 12:30:17,115 WARN
org.apache.kafka.clients.admin.AdminClientConfig [] - The
configuration 'auto.offset.reset' was supplied but isn't a known config.
It seems that they are bogus.
Regards Hans-Peter
Properties sinkkafkaProps = new Properties();
sinkkafkaProps.setProperty("ssl.truststore.type", outputTrustStoreType);
sinkkafkaProps.setProperty("ssl.truststore.location", outputTrustStoreLocation);
sinkkafkaProps.setProperty("ssl.truststore.password", outputTrustStorePassword);
sinkkafkaProps.setProperty("security.protocol", outputSecurityProtocol);
sinkkafkaProps.setProperty("max.request.size", maxRequestSize);
sinkkafkaProps.setProperty("partition.discovery.interval.ms",
partitionDiscoveryIntervalMs);
sinkkafkaProps.setProperty("commit.offsets.on.checkpoint",
commitOffsetsOnCheckpoint);
if (kafkaOutputDisabled.equals("false")) {
KafkaSink<String> kSink = KafkaSink.<String>builder()
.setBootstrapServers(outputBrokers)
.setKafkaProducerConfig(sinkkafkaProps)
.setRecordSerializer(KafkaRecordSerializationSchema.builder()
.setTopic(kafkaOutputTopic)
.setValueSerializationSchema(new SimpleStringSchema())
.build()
)
.setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
.build();
Op do 17 mrt. 2022 om 17:29 schreef Matthias Pohl <matth...@ververica.com>:
> Could you share more details on what's not working? Is the
> ssl.trustore.location accessible from the Flink nodes?
>
> Matthias
>
> On Thu, Mar 17, 2022 at 4:00 PM HG <hanspeter.sl...@gmail.com> wrote:
>
>> Hi all,
>> I am probably not the smartest but I cannot find how to set
>> ssl-properties for a Kafka Sink.
>> My assumption was that it would be just like the Kafka Consumer
>>
>> KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder()
>> .setProperties(kafkaProps)
>> .setProperty("ssl.truststore.type", trustStoreType)
>> .setProperty("ssl.truststore.password", trustStorePassword)
>> .setProperty("ssl.truststore.location", trustStoreLocation)
>> .setProperty("security.protocol", securityProtocol)
>> .setProperty("partition.discovery.interval.ms",
>> partitionDiscoveryIntervalMs)
>> .setProperty("commit.offsets.on.checkpoint",
>> commitOffsetsOnCheckpoint)
>> .setGroupId(inputGroupId)
>> .setClientIdPrefix(clientId)
>> .setTopics(kafkaInputTopic)
>> .setDeserializer(KafkaRecordDeserializationSchema.of(new
>> JSONKeyValueDeserializationSchema(fetchMetadata)))
>>
>> .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
>> .build();
>>
>>
>> But that seems not to be the case.
>>
>> Any quick pointers?
>>
>> Regards Hans-Peter
>>
>
