Docs Cloud Redpanda Connect Components Catalog Inputs legacy_redpanda_migrator legacy_redpanda_migrator Type: Input ▼ InputOutput Available in: Cloud, Self-Managed Deprecated in 4.67.5 This component is deprecated and will be removed in the next major version release. Please consider moving onto the unified redpanda_migrator input and redpanda_migrator output components. For migration instructions, see Migrate to the Unified Redpanda Migrator. Use this connector in conjunction with the legacy_redpanda_migrator output to migrate topics between Apache Kafka brokers. The legacy_redpanda_migrator input uses the Franz Kafka client library. Common Advanced inputs: label: "" legacy_redpanda_migrator: seed_brokers: [] # No default (required) topics: [] # No default (required) regexp_topics: false transaction_isolation_level: read_uncommitted consumer_group: "" # No default (optional) auto_replay_nacks: true inputs: label: "" legacy_redpanda_migrator: seed_brokers: [] # No default (required) client_id: benthos tls: enabled: false skip_cert_verify: false enable_renegotiation: false root_cas: "" root_cas_file: "" client_certs: cert: "" key: "" cert_file: "" key_file: "" password: "" sasl: mechanism: "" # No default (required) username: "" password: "" token: "" extensions: "" # No default (optional) aws: region: "" # No default (optional) endpoint: "" # No default (optional) credentials: profile: "" # No default (optional) id: "" # No default (optional) secret: "" # No default (optional) token: "" # No default (optional) from_ec2_role: "" # No default (optional) role: "" # No default (optional) role_external_id: "" # No default (optional) metadata_max_age: 5m request_timeout_overhead: 10s conn_idle_timeout: 20s topics: [] # No default (required) regexp_topics: false rack_id: "" instance_id: "" rebalance_timeout: 45s session_timeout: 1m heartbeat_interval: 3s start_offset: earliest fetch_max_bytes: 50MiB fetch_max_wait: 5s fetch_min_bytes: 1B fetch_max_partition_bytes: 1MiB transaction_isolation_level: read_uncommitted consumer_group: "" # No default (optional) commit_period: 5s partition_buffer_bytes: 1MB topic_lag_refresh_period: 5s max_yield_batch_bytes: 32KB auto_replay_nacks: true timely_nacks_maximum_wait: "" # No default (optional) The legacy_redpanda_migrator input: Reads a batch of messages from a Kafka broker. Waits for the legacy_redpanda_migrator output to acknowledge the writes before updating the Kafka consumer group offset. Provides the same delivery guarantees and ordering semantics as the redpanda input. Specify a consumer group to make this input consume one or more topics and automatically balance the topic partitions across any other connected clients with the same consumer group. Otherwise, topics are consumed in their entirety or with explicit partitions. Metrics This input emits an input_redpanda_migrator_lag metric with topic and partition labels for each consumed topic. The metric records the number of produced messages that remain to be read from each topic/partition pair by the specified consumer group. Metadata This input adds the following metadata fields to each message: - kafka_key - kafka_topic - kafka_partition - kafka_offset - kafka_lag - kafka_timestamp_ms - kafka_timestamp_unix - All record headers Fields auto_replay_nacks Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Type: bool Default: true client_id An identifier for the client connection. Type: string Default: benthos commit_period The period of time between each commit of the current partition offsets. Offsets are always committed during shutdown. Type: string Default: 5s conn_idle_timeout The rough amount of time to allow connections to idle before they are closed. Type: string Default: 20s consumer_group An optional consumer group to consume as. When specified the partitions of specified topics are automatically distributed across consumers sharing a consumer group, and partition offsets are automatically committed and resumed under this name. Consumer groups are not supported when specifying explicit partitions to consume from in the topics field. Type: string fetch_max_bytes Sets the maximum amount of bytes a broker will try to send during a fetch. Note that brokers may not obey this limit if it has records larger than this limit. This is the equivalent to the Java fetch.max.bytes setting. Type: string Default: 50MiB fetch_max_partition_bytes Sets the maximum amount of bytes that will be consumed for a single partition in a fetch request. Note that if a single batch is larger than this number, that batch will still be returned so the client can make progress. This is the equivalent to the Java fetch.max.partition.bytes setting. Type: string Default: 1MiB fetch_max_wait Sets the maximum amount of time a broker will wait for a fetch response to hit the minimum number of required bytes. This is the equivalent to the Java fetch.max.wait.ms setting. Type: string Default: 5s fetch_min_bytes Sets the minimum amount of bytes a broker will try to send during a fetch. This is the equivalent to the Java fetch.min.bytes setting. Type: string Default: 1B heartbeat_interval When using a consumer group, heartbeat_interval sets how long a group member goes between heartbeats to Kafka. Kafka uses heartbeats to ensure that a group member’s sesion stays active. This value should be no higher than 1/3rd of the session_timeout. This is equivalent to the Java heartbeat.interval.ms setting. Type: string Default: 3s instance_id When using a consumer group, an instance ID specifies the groups static membership, which can prevent rebalances during reconnects. When using a instance ID the client does NOT leave the group when closing. To actually leave the group one must use an external admin command to leave the group on behalf of this instance ID. This ID must be unique per consumer within the group. Type: string Default: "" max_yield_batch_bytes The maximum size (in bytes) for each batch yielded by this input. When routed to a redpanda output without modification this would roughly translate to the batch.bytes config field of a traditional producer. Type: string Default: 32KB metadata_max_age The maximum age of metadata before it is refreshed. This interval also controls how frequently regex topic patterns are re-evaluated to discover new matching topics. Type: string Default: 5m partition_buffer_bytes A buffer size (in bytes) for each consumed partition, allowing records to be queued internally before flushing. Increasing this may improve throughput at the cost of higher memory utilisation. Note that each buffer can grow slightly beyond this value. Type: string Default: 1MB rack_id A rack specifies where the client is physically located and changes fetch requests to consume from the closest replica as opposed to the leader replica. Type: string Default: "" rebalance_timeout When using a consumer group, rebalance_timeout sets how long group members are allowed to take when a rebalance has begun. This timeout is how long all members are allowed to complete work and commit offsets, minus the time it took to detect the rebalance (from a heartbeat). Type: string Default: 45s regexp_topics Whether listed topics should be interpreted as regular expression patterns for matching multiple topics. When enabled, the client will periodically refresh the list of matching topics based on the metadata_max_age interval. When topics are specified with explicit partitions this field must remain set to false. Type: bool Default: false request_timeout_overhead The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields. Type: string Default: 10s sasl[] Specify one or more methods of SASL authentication. SASL is tried in order; if the broker supports the first mechanism, all connections will use that mechanism. If the first mechanism fails, the client will pick the first supported mechanism. If the broker does not support any client mechanisms, connections will fail. Type: object # Examples: sasl: - mechanism: SCRAM-SHA-512 password: bar username: foo sasl[].aws Contains AWS specific fields for when the mechanism is set to AWS_MSK_IAM. Type: object sasl[].aws.credentials Optional manual configuration of AWS credentials to use. More information can be found in Amazon Web Services. Type: object sasl[].aws.credentials.from_ec2_role Use the credentials of a host EC2 machine configured to assume an IAM role associated with the instance. Type: bool sasl[].aws.credentials.id The ID of credentials to use. Type: string sasl[].aws.credentials.profile A profile from ~/.aws/credentials to use. Type: string sasl[].aws.credentials.role A role ARN to assume. Type: string sasl[].aws.credentials.role_external_id An external ID to provide when assuming a role. Type: string sasl[].aws.credentials.secret The secret for the credentials being used. This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration. Type: string sasl[].aws.credentials.token The token for the credentials being used, required when using short term credentials. Type: string sasl[].aws.endpoint Allows you to specify a custom endpoint for the AWS API. Type: string sasl[].aws.region The AWS region to target. Type: string sasl[].extensions Key/value pairs to add to OAUTHBEARER authentication requests. Type: string sasl[].mechanism The SASL mechanism to use. Type: string Option Summary AWS_MSK_IAM AWS IAM based authentication as specified by the 'aws-msk-iam-auth' java library. OAUTHBEARER OAuth Bearer based authentication. PLAIN Plain text authentication. SCRAM-SHA-256 SCRAM based authentication as specified in RFC5802. SCRAM-SHA-512 SCRAM based authentication as specified in RFC5802. none Disable sasl authentication sasl[].password A password to provide for PLAIN or SCRAM-* authentication. This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration. Type: string Default: "" sasl[].token The token to use for a single session’s OAUTHBEARER authentication. Type: string Default: "" sasl[].username A username to provide for PLAIN or SCRAM-* authentication. Type: string Default: "" seed_brokers[] A list of broker addresses to connect to in order to establish connections. If an item of the list contains commas it will be expanded into multiple addresses. Type: array # Examples: seed_brokers: - "localhost:9092" - "foo:9092" - "bar:9092" - "foo:9092,bar:9092" session_timeout When using a consumer group, session_timeout sets how long a member in hte group can go between heartbeats. If a member does not heartbeat in this timeout, the broker will remove the member from the group and initiate a rebalance. Type: string Default: 1m start_offset Sets the offset to start consuming from, or if OffsetOutOfRange is seen while fetching, to restart consuming from. Type: string Default: earliest Option Summary committed Prevents consuming a partition in a group if the partition has no prior commits. Corresponds to Kafka’s auto.offset.reset=none option earliest Start from the earliest offset. Corresponds to Kafka’s auto.offset.reset=earliest option. latest Start from the latest offset. Corresponds to Kafka’s auto.offset.reset=latest option. timely_nacks_maximum_wait EXPERIMENTAL: Specify a maximum period of time in which each message can be consumed and awaiting either acknowledgement or rejection before rejection is instead forced. This can be useful for avoiding situations where certain downstream components can result in blocked confirmation of delivery that exceeds SLAs. Type: string tls Custom TLS settings can be used to override system defaults. Type: object tls.client_certs[] A list of client certificates to use. For each certificate either the fields cert and key, or cert_file and key_file should be specified, but not both. Type: object Default: [] # Examples: client_certs: - cert: foo key: bar - cert_file: ./example.pem key_file: ./example.key tls.client_certs[].cert A plain text certificate to use. Type: string Default: "" tls.client_certs[].cert_file The path of a certificate to use. Type: string Default: "" tls.client_certs[].key A plain text certificate key to use. This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration. Type: string Default: "" tls.client_certs[].key_file The path of a certificate key to use. Type: string Default: "" tls.client_certs[].password A plain text password for when the private key is password encrypted in PKCS#1 or PKCS#8 format. The obsolete pbeWithMD5AndDES-CBC algorithm is not supported for the PKCS#8 format. Because the obsolete pbeWithMD5AndDES-CBC algorithm does not authenticate the ciphertext, it is vulnerable to padding oracle attacks that can let an attacker recover the plaintext. This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration. Type: string Default: "" # Examples: password: foo password: ${KEY_PASSWORD} tls.enable_renegotiation Whether to allow the remote server to repeatedly request renegotiation. Enable this option if you’re seeing the error message local error: tls: no renegotiation. Type: bool Default: false tls.enabled Whether custom TLS settings are enabled. Type: bool Default: false tls.root_cas An optional root certificate authority to use. This is a string, representing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate. This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration. Type: string Default: "" # Examples: root_cas: |- -----BEGIN CERTIFICATE----- ... -----END CERTIFICATE----- tls.root_cas_file An optional path of a root certificate authority file to use. This is a file, often with a .pem extension, containing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate. Type: string Default: "" # Examples: root_cas_file: ./root_cas.pem tls.skip_cert_verify Whether to skip server side certificate verification. Type: bool Default: false topic_lag_refresh_period The period of time between each topic lag refresh cycle. Type: string Default: 5s topics[] A list of topics to consume from. Multiple comma separated topics can be listed in a single element. When a consumer_group is specified partitions are automatically distributed across consumers of a topic, otherwise all partitions are consumed. Alternatively, it’s possible to specify explicit partitions to consume from with a colon after the topic name, e.g. foo:0 would consume the partition 0 of the topic foo. This syntax supports ranges, e.g. foo:0-10 would consume partitions 0 through to 10 inclusive. Finally, it’s also possible to specify an explicit offset to consume from by adding another colon after the partition, e.g. foo:0:10 would consume the partition 0 of the topic foo starting from the offset 10. If the offset is not present (or remains unspecified) then the field start_from_oldest determines which offset to start from. Type: array # Examples: topics: - foo - bar - things.* - "foo,bar" - "foo:0" - "bar:1" - "bar:3" - "foo:0,bar:1,bar:3" - "foo:0-5" transaction_isolation_level The transaction isolation level Type: string Default: read_uncommitted Option Summary read_committed If set, only committed transactional records are processed. read_uncommitted If set, then uncommitted records are processed. Back to top × Simple online edits For simple changes, such as fixing a typo, you can edit the content directly on GitHub. Edit on GitHub Or, open an issue to let us know about something that you want us to change. Open an issue Contribution guide For extensive content updates, or if you prefer to work locally, read our contribution guide . Was this helpful? thumb_up thumb_down group Ask in the community mail Share your feedback group_add Make a contribution 🎉 Thanks for your feedback! kafka_franz legacy_redpanda_migrator_offsets