Merge branch 'main' into wkx/cdc-source-wait-timeout

.github/workflows/cherry-pick-to-release-branch.yml

@@ -6,8 +6,8 @@ on:
     types: ["closed", "labeled"]
 
 jobs:
-  release_pull_request_1_8:
-    if: "contains(github.event.pull_request.labels.*.name, 'need-cherry-pick-release-1.8')  &&  github.event.pull_request.merged == true"
+  release_pull_request_1_9:
+    if: "contains(github.event.pull_request.labels.*.name, 'need-cherry-pick-release-1.9')  &&  github.event.pull_request.merged == true"
     runs-on: ubuntu-latest
     name: release_pull_request
     steps:
@@ -16,14 +16,14 @@ jobs:
       - name: Create PR to branch
         uses: risingwavelabs/github-action-cherry-pick@master
         with:
-          pr_branch: 'release-1.8'
+          pr_branch: 'release-1.9'
           pr_labels: 'cherry-pick'
-          pr_body: ${{ format('Cherry picking \#{0} onto branch release-1.8', github.event.number) }}
+          pr_body: ${{ format('Cherry picking \#{0} onto branch release-1.9', github.event.number) }}
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
-  release_pull_request_1_7_standalone:
-    if: "contains(github.event.pull_request.labels.*.name, 'need-cherry-pick-release-1.7-standalone')  &&  github.event.pull_request.merged == true"
+  release_pull_request_1_8:
+    if: "contains(github.event.pull_request.labels.*.name, 'need-cherry-pick-release-1.8')  &&  github.event.pull_request.merged == true"
     runs-on: ubuntu-latest
     name: release_pull_request
     steps:
@@ -32,24 +32,9 @@ jobs:
       - name: Create PR to branch
         uses: risingwavelabs/github-action-cherry-pick@master
         with:
-          pr_branch: 'release-1.7.0-standalone'
+          pr_branch: 'release-1.8'
           pr_labels: 'cherry-pick'
-          pr_body: ${{ format('Cherry picking \#{0} onto branch release-1.7.0-standalone', github.event.number) }}
+          pr_body: ${{ format('Cherry picking \#{0} onto branch release-1.8', github.event.number) }}
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
-  release_pull_request_1_7:
-    if: "contains(github.event.pull_request.labels.*.name, 'need-cherry-pick-release-1.7')  &&  github.event.pull_request.merged == true"
-    runs-on: ubuntu-latest
-    name: release_pull_request
-    steps:
-      - name: checkout
-        uses: actions/checkout@v1
-      - name: Create PR to branch
-        uses: risingwavelabs/github-action-cherry-pick@master
-        with:
-          pr_branch: 'release-1.7'
-          pr_labels: 'cherry-pick'
-          pr_body: ${{ format('Cherry picking \#{0} onto branch release-1.7', github.event.number) }}
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

Makefile.toml

@@ -697,33 +697,19 @@ wait_kafka_exit() {
   done
 }
 
-wait_zookeeper_exit() {
-  # Follow zookeeper-server-stop.sh
-  while [[ -n "$(ps ax | grep java | grep -i QuorumPeerMain | grep -v grep | awk '{print $1}')" ]]; do
-    echo "Waiting for zookeeper to exit"
-    sleep 1
-  done
-}
-
 kill_kafka() {
   ${PREFIX_BIN}/kafka/bin/kafka-server-stop.sh
   wait_kafka_exit
 }
 
-kill_zookeeper() {
-  ${PREFIX_BIN}/kafka/bin/zookeeper-server-stop.sh
-  wait_zookeeper_exit
-}
-
 if ! ${TMUX} ls &>/dev/null ; then
   echo "No risedev cluster to kill. Exiting..."
   exit 0
 fi
 
-# Kill other components
+# Kill other components with Ctrl+C/Ctrl+D
 ${TMUX} list-windows -F "#{window_name} #{pane_id}" \
-| grep -v 'kafka' \
-| grep -v 'zookeeper' \
+| grep --invert-match --extended-regexp '(kafka)' \
 | awk '{ print $2 }' \
 | xargs -I {} ${TMUX} send-keys -t {} C-c C-d
 
@@ -734,17 +720,11 @@ if [[ -n ${containers} ]]; then
   docker stop ${containers}
 fi
 
-# Kill kafka and zookeeper
+# Kill kafka cleanly. Ctrl+C will lose data.
 if [[ -n $(${TMUX} list-windows | grep kafka) ]];
 then
   echo "kill kafka"
   kill_kafka || true
-
-  echo "kill zookeeper"
-  kill_zookeeper || true
-
-  # Kill their ${TMUX} sessions
-  ${TMUX} list-windows -t risedev -F "#{pane_id}" | xargs -I {} ${TMUX} send-keys -t {} C-c C-d
 fi
 
 ${TMUX} kill-server

backwards-compat-tests/scripts/run_local.sh

@@ -23,16 +23,17 @@ full-without-monitoring:
     - use: compute-node
     - use: frontend
     - use: compactor
-    - use: zookeeper
     - use: kafka
+      user-managed: true
+      address: message_queue
+      port: 29092
 EOF
 
 cat <<EOF > risedev-components.user.env
 RISEDEV_CONFIGURED=false
 
 ENABLE_MINIO=true
 ENABLE_ETCD=true
-ENABLE_KAFKA=true
 
 # Fetch risingwave binary from release.
 ENABLE_BUILD_RUST=true
@@ -65,4 +66,4 @@ main() {
   validate_new_cluster "$NEW_VERSION"
 }
 
-main
+main

backwards-compat-tests/scripts/utils.sh

@@ -19,36 +19,9 @@ RECOVERY_DURATION=20
 
 # Setup test directory
 TEST_DIR=.risingwave/backwards-compat-tests/
-KAFKA_PATH=.risingwave/bin/kafka
 mkdir -p $TEST_DIR
 cp -r backwards-compat-tests/slt/* $TEST_DIR
 
-wait_kafka_exit() {
-  # Follow kafka-server-stop.sh
-  while [[ -n "$(ps ax | grep ' kafka\.Kafka ' | grep java | grep -v grep | awk '{print $1}')" ]]; do
-    echo "Waiting for kafka to exit"
-    sleep 1
-  done
-}
-
-wait_zookeeper_exit() {
-  # Follow zookeeper-server-stop.sh
-  while [[ -n "$(ps ax | grep java | grep -i QuorumPeerMain | grep -v grep | awk '{print $1}')" ]]; do
-    echo "Waiting for zookeeper to exit"
-    sleep 1
-  done
-}
-
-kill_kafka() {
-  $KAFKA_PATH/bin/kafka-server-stop.sh
-  wait_kafka_exit
-}
-
-kill_zookeeper() {
-  $KAFKA_PATH/bin/zookeeper-server-stop.sh
-  wait_zookeeper_exit
-}
-
 wait_for_process() {
   process_name="$1"
 
@@ -77,24 +50,9 @@ kill_cluster() {
 
   # Kill other components
   $TMUX list-windows -t risedev -F "#{window_name} #{pane_id}" |
-    grep -v 'kafka' |
-    grep -v 'zookeeper' |
     awk '{ print $2 }' |
     xargs -I {} $TMUX send-keys -t {} C-c C-d
 
-  set +e
-  if [[ -n $($TMUX list-windows -t risedev | grep kafka) ]]; then
-    echo "kill kafka"
-    kill_kafka
-
-    echo "kill zookeeper"
-    kill_zookeeper
-
-    # Kill their tmux sessions
-    $TMUX list-windows -t risedev -F "#{pane_id}" | xargs -I {} $TMUX send-keys -t {} C-c C-d
-  fi
-  set -e
-
   $TMUX kill-server
   test $? -eq 0 || {
     echo "Failed to stop all RiseDev components."
@@ -120,18 +78,16 @@ check_version() {
 }
 
 create_kafka_topic() {
-  "$KAFKA_PATH"/bin/kafka-topics.sh \
-    --create \
-    --topic backwards_compat_test_kafka_source --bootstrap-server localhost:29092
+  RPK_BROKERS=message_queue:29092 \
+  rpk topic create backwards_compat_test_kafka_source
 }
 
 insert_json_kafka() {
   local JSON=$1
-  echo "$JSON" | "$KAFKA_PATH"/bin/kafka-console-producer.sh \
-    --topic backwards_compat_test_kafka_source \
-    --bootstrap-server localhost:29092 \
-    --property "parse.key=true" \
-    --property "key.separator=,"
+
+  echo "$JSON" | \
+  RPK_BROKERS=message_queue:29092 \
+  rpk topic produce backwards_compat_test_kafka_source -f "%k,%v"
 }
 
 seed_json_kafka() {

backwards-compat-tests/slt/kafka/invalid_options/seed.slt

@@ -14,7 +14,7 @@ CREATE SOURCE IF NOT EXISTS kafka_source_with_invalid_option
 WITH (
  connector='kafka',
  topic='backwards_compat_test_kafka_source',
- properties.bootstrap.server='localhost:29092',
+ properties.bootstrap.server='message_queue:29092',
  scan.startup.mode='earliest',
  invalid_option='oops'
 ) FORMAT PLAIN ENCODE JSON;

backwards-compat-tests/slt/kafka/seed.slt

@@ -11,7 +11,7 @@ CREATE SOURCE IF NOT EXISTS kafka_source
 WITH (
  connector='kafka',
  topic='backwards_compat_test_kafka_source',
- properties.bootstrap.server='localhost:29092',
+ properties.bootstrap.server='message_queue:29092',
  scan.startup.mode='earliest',
 ) FORMAT PLAIN ENCODE JSON;
 

backwards-compat-tests/slt/kafka/upsert/deprecate_upsert.slt

@@ -11,6 +11,6 @@ CREATE TABLE IF NOT EXISTS kafka_table
 WITH (
  connector='kafka',
  topic='backwards_compat_test_kafka_source',
- properties.bootstrap.server='localhost:29092',
+ properties.bootstrap.server='message_queue:29092',
  scan.startup.mode='earliest',
-) FORMAT UPSERT ENCODE JSON;
+) FORMAT UPSERT ENCODE JSON;

backwards-compat-tests/slt/kafka/upsert/include_key_as.slt

@@ -13,6 +13,6 @@ INCLUDE key as _rw_key
 WITH (
  connector='kafka',
  topic='backwards_compat_test_kafka_source',
- properties.bootstrap.server='localhost:29092',
+ properties.bootstrap.server='message_queue:29092',
  scan.startup.mode='earliest',
-) FORMAT UPSERT ENCODE JSON;
+) FORMAT UPSERT ENCODE JSON;