# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
bootstrap.servers={{ kafka.bootstrap_servers(kafka.security_config.security_protocol) }}
{{ kafka.security_config.client_config().props() }}
{{ kafka.security_config.client_config().props("producer.") }}
{{ kafka.security_config.client_config().props("consumer.") }}
group.id={{ group|default("connect-cluster") }}
exactly.once.source.support={{ EXACTLY_ONCE_SOURCE_SUPPORT|default("disabled") }}
{% if EXACTLY_ONCE_SOURCE_SUPPORT is defined and EXACTLY_ONCE_SOURCE_SUPPORT == "enabled" %}
# Reduce transaction timeouts so tests that kill workers don't need to wait as long to recover
producer.transaction.timeout.ms = 10000
{% endif %}
connect.protocol={{ CONNECT_PROTOCOL|default("sessioned") }}
scheduled.rebalance.max.delay.ms={{ SCHEDULED_REBALANCE_MAX_DELAY_MS|default(60000) }}
key.converter={{ key_converter|default("org.apache.kafka.connect.json.JsonConverter") }}
value.converter={{ value_converter|default("org.apache.kafka.connect.json.JsonConverter") }}
{% if key_converter is not defined or key_converter.endswith("JsonConverter") %}
key.converter.schemas.enable={{ schemas|default(True)|string|lower }}
{% endif %}
{% if value_converter is not defined or value_converter.endswith("JsonConverter") %}
value.converter.schemas.enable={{ schemas|default(True)|string|lower }}
{% endif %}
offset.storage.topic={{ OFFSETS_TOPIC }}
offset.storage.replication.factor={{ OFFSETS_REPLICATION_FACTOR }}
offset.storage.partitions={{ OFFSETS_PARTITIONS }}
config.storage.topic={{ CONFIG_TOPIC }}
config.storage.replication.factor={{ CONFIG_REPLICATION_FACTOR }}
status.storage.topic={{ STATUS_TOPIC }}
status.storage.replication.factor={{ STATUS_REPLICATION_FACTOR }}
status.storage.partitions={{ STATUS_PARTITIONS }}
# Make sure data gets flushed frequently so tests don't have to wait to ensure they see data in output systems
offset.flush.interval.ms=5000
rest.advertised.host.name = {{ node.account.hostname }}
# Reduce the admin client request timeouts so that we don't wait the default 120 sec before failing to connect the admin client
request.timeout.ms=30000
# Allow connector configs to use externalized config values of the form:
# ${file:/mnt/connect/connect-external-configs.properties:topic.external}
#
config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider
{% if PLUGIN_PATH is defined %}
plugin.path={{ PLUGIN_PATH }}
{% endif %}
plugin.discovery={{ PLUGIN_DISCOVERY|default("service_load") }}