Kafka upgrade version

README.md

@@ -44,14 +44,14 @@ The amq_streams collection also depends on the following python packages to be p
 A requirement file is provided to install:
 
     pip install -r requirements.txt
-
+<!--end galaxy_download -->
 ### Build and install locally
 
 Clone the repository, checkout the tag you want to build, or pick the main branch for the development version; then:
 
     ansible-galaxy collection build .
     ansible-galaxy collection install middleware_automation-amq_streams-*.tar.gz
-<!--end galaxy_download -->
+
 ## Usage
 
 ### Install Playbook

roles/amq_streams_broker/README.md

@@ -61,6 +61,8 @@ broker4
 |`amq_streams_broker_transaction_state_log_min_isr` |  | `1` |
 |`amq_streams_broker_log_retention_hours` |  | `168` |
 |`amq_streams_broker_log_retention_check_interval_ms` |  | `300000` |
+|`amq_streams_broker_log_message_format_version` | Log message format version. Required for Kafka upgrade. |  |
+|`amq_streams_broker_inter_broker_protocol_version` | Inter broker protocol version. Required for Kafka upgrade. |  |
 |`amq_streams_broker_zookeeper_connection_timeout_ms` |  | `18000` |
 |`amq_streams_broker_group_initial_rebalance_delay_ms` |  | `0` |
 |`amq_streams_broker_properties_template` |  | `templates/server.properties.j2` |
@@ -81,7 +83,7 @@ broker4
 |`amq_streams_broker_auth_listeners` | Default list of authenticated listeners | `PLAINTEXT:PLAINTEXT` |
 |`amq_streams_broker_auth_sasl_mechanisms` | Default list of authenticated SASL mechanism | `PLAIN` |
 |`amq_streams_broker_inventory_group` | Identify the group of broker nodes | `groups['brokers']` |
-|`amq_streams_broker_broker_id` | Identify the broker with specific id in the inventory
+|`amq_streams_broker_broker_id` | Identify the broker with specific id in the inventory. | |
 |`amq_streams_broker_topics` | List of topics to create. Each topics requires the `name` property, and optionally the `partitions` and `replication_factor`. | |
 
 ## Role Variables
@@ -257,7 +259,6 @@ To manage SCRAM users, the role includes the following tasks:
 * Describe
 * Delete
 
-
 The role uses the `amq_streams_broker_topics` variable to identify the list of topics
 to be managed by the role for each stage of the life cycle.
 
@@ -456,6 +457,87 @@ Example of definition:
     amq_streams_broker_admin_password: password
 ```
 
+## Kafka Upgrading
+
+Kafka upgrading requires a coordinated exercise between the actions to use the latest version of Apache Kafka, and the different applications
+(consumer and producers). This process requires first to use the lates version of Apache Kafka, maintaining the same log message format and
+inter-broker protocol. Once the platform is using the new binaries of the latest version, then an upgrade of the log message format and
+inter-broker protocol can be done. This upgrade process must require finally code changes to client applications (consumers and producers) to
+align the log message format to the newer version of the Kafka brokers.
+
+The log message format version is managed by the `log.message.format.version` property of Kafka. This property is managed by
+the `amq_streams_broker_log_message_format_version` role variable.
+
+The inter-broker protocol version is managed by the `inter.broker.protocol.version` property of Kafka. This property is managed by the
+`amq_streams_broker_inter_broker_protocol_version` role variable.
+
+The execution of the playbook during an upgrading process must be done following a linear strategy of Ansible. The variable `serial` must be
+defined with the value `1`.
+
+**NOTE**: The upgrade process will deploy the latest version of Apache Kafka in a new folder, without any change in the previous Kafka deployment. The
+service units will be updated to use the new location of the binaries installed. This process allows a quickly rollback in case of emergency.
+
+Upgrading a Kafka cluster by this collection requires to follow this playbook sequence execution (Example to upgrade from Kafka 3.5 to Kafka 3.6):
+
+1. (Optional) Run the collection to set up the log message format and inter-broker protocol version if they are not already defined. Example of definition:
+
+```yaml
+---
+- name: "Ansible Playbook to set up the log message format and inter-broker protocol version"
+  hosts: all
+  serial: 1
+  vars:
+    # Kafka version
+    amq_streams_common_scala_version: 2.13
+    amq_streams_common_product_version: 3.5.0
+    # Log message format version
+    amq_streams_broker_log_message_format_version: 3.5
+    # Inter-broker protocol version
+    amq_streams_broker_inter_broker_protocol_version: 3.5
+```
+
+2. Run the collection to deploy the latest version of Apache Kafka enabling the `amq_streams_common_product_upgrade` variable. Log message format and inter-broker protocol
+version are not changed in this execution. Example of definition:
+
+```yaml
+---
+- name: "Ansible Playbook to upgrade using the binaries from the new version"
+  hosts: all
+  serial: 1
+  vars:
+    # Kafka version
+    amq_streams_common_scala_version: 2.13
+    amq_streams_common_product_version: 3.6.0
+    amq_streams_common_product_upgrade: true
+    # Log message format version
+    amq_streams_broker_log_message_format_version: 3.5
+    # Inter-broker protocol version
+    amq_streams_broker_inter_broker_protocol_version: 3.5
+```
+
+3. Run the collection to upgrade the log message format and inter-broker protocol version. This execution must be coordinated with the applications when they
+are ready to use the new versions.
+
+```yaml
+---
+- name: "Ansible Playbook to upgrade the log message format and inter-broker protocol version for the new version upgraded"
+  hosts: all
+  serial: 1
+  vars:
+    # Kafka version
+    amq_streams_common_scala_version: 2.13
+    amq_streams_common_product_version: 3.6.0
+    amq_streams_common_product_upgrade: true
+    # Log message format version
+    amq_streams_broker_log_message_format_version: 3.6
+    # Inter-broker protocol version
+    amq_streams_broker_inter_broker_protocol_version: 3.6
+```
+
+**NOTE:** Once the Kafka upgrade is finished, the variable `amq_streams_common_product_upgrade` must be removed from any playbook, or change to `false`.
+
+[Apache Kafka upgrading reference](https://kafka.apache.org/documentation/#upgrade_3_6_0)
+
 ## License
 
 Apache License v2.0 or later

roles/amq_streams_broker/templates/server.properties.j2

@@ -25,7 +25,6 @@
 # The id of the broker. This must be set to a unique integer for each broker.
 broker.id={{ amq_streams_broker_broker_id | default(amq_streams_broker_inventory_group.index(inventory_hostname)) }}
 
-
 ############################# Socket Server Settings #############################
 
 # The address the socket server listens on. If not configured, the host name will be equal to the value of
@@ -184,8 +183,19 @@ zookeeper.session.timeout.ms={{ amq_streams_broker_zookeeper_session_timeout_ms
 # However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
 group.initial.rebalance.delay.ms={{ amq_streams_broker_group_initial_rebalance_delay_ms }}
 
+############################# Kafka Authorization #############################
+
 {%if amq_streams_broker_acl_enabled is defined and amq_streams_broker_acl_enabled %}
 authorizer.class.name=kafka.security.authorizer.AclAuthorizer
 super.users=User:admin
 allow.everyone.if.no.acl.found=true
 {% endif %}
+
+############################# Log Message format and Interbroker protocol #############################
+
+{% if amq_streams_broker_log_message_format_version is defined %}
+log.message.format.version: {{ amq_streams_broker_log_message_format_version }}
+{% endif %}
+{% if amq_streams_broker_inter_broker_protocol_version is defined %}
+inter.broker.protocol.version: {{ amq_streams_broker_inter_broker_protocol_version }}
+{% endif %}

roles/amq_streams_common/README.md

@@ -6,7 +6,8 @@ Common tasks for the collection.
 
 | Variable | Description | Default |
 |:---------|:------------|:--------|
-|`amq_streams_common_product_version` | Kafka version | `3.4.0` |
+|`amq_streams_common_product_upgrade` | Enable a product upgrade | `false` |
+|`amq_streams_common_product_version` | Kafka version | `3.6.0` |
 |`amq_streams_common_scala_version` | Scala version | `2.13` |
 |`amq_streams_common_version` | Combination version | `{{ amq_streams_common_scala_version }}-{{ amq_streams_common_product_version }}` |
 |`amq_streams_common_archive_file` | Kafka binary package | `kafka_{{ amq_streams_common_version }}.tgz` |
@@ -30,7 +31,6 @@ The following are a set of required variables for the role:
 | Variable | Description | Required |
 |:---------|:------------|:---------|
 
-
 ## Offline installation
 
 Performing an offline installation is possible by:
@@ -47,4 +47,3 @@ Apache License v2.0 or later
 * [Romain Pelisse](https://github.com/rpelisse)
 * [Guido Grazioli](https://github.com/guidograzioli)
 * [Roman Martin](https://github.com/rmarting)
-

roles/amq_streams_common/defaults/main.yml

@@ -1,4 +1,5 @@
 ---
+amq_streams_common_product_upgrade: false
 amq_streams_common_product_version: 3.6.0
 amq_streams_common_rhn_product_version: 2.5.1
 amq_streams_common_rhn_home_dir: kafka_2.13-3.5.0.redhat-00014

roles/amq_streams_common/tasks/systemd.yml

@@ -74,3 +74,12 @@
     enabled: yes
     state: started
   become: yes
+
+- name: "Ensure {{ server_name }} is restarted after a product upgrade."
+  ansible.builtin.set_fact:
+    amq_streams_common_product_upgrade_status: "Upgrading {{ server_name }}"
+  when:
+    - amq_streams_common_product_upgrade is defined and amq_streams_common_product_upgrade
+  notify:
+    - "Restart {{ server_name }}"
+  become: yes