יצירת אשכול של שירות מנוהל של Google Cloud ל-Apache Kafka

אשכול של שירות מנוהל ל-Apache Kafka מספק סביבה לאחסון ולעיבוד של זרמי הודעות שמחולקים לנושאים.

כדי ליצור אשכול, אתם יכולים להשתמש במסוף Google Cloud , ב-Google Cloud CLI, בספריית הלקוח או ב-Managed Kafka API. אי אפשר להשתמש ב-API של Apache Kafka בקוד פתוח כדי ליצור אשכול.

לפני שמתחילים

ודאו שאתם מכירים את הנושאים הבאים:

• איך פועל השירות המנוהל ל-Apache Kafka

• מושגים ברשת שקשורים לשירות המנוהל ל-Apache Kafka

תפקידים והרשאות נדרשים

כדי לקבל את ההרשאות שנדרשות ליצירת אשכול, צריך לבקש מהאדמין להקצות לכם ב-IAM את התפקיד עריכת אשכולות מנוהלים של Kafka (roles/managedkafka.clusterEditor) בפרויקט. כדי לקרוא הסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

התפקיד המוגדר מראש הזה כולל את ההרשאה managedkafka.clusters.create, שנדרשת כדי ליצור אשכול.

יכול להיות שתוכלו לקבל את ההרשאה הזו גם בתפקידים בהתאמה אישית או בתפקידים מוגדרים מראש.

הרשאות של רשתות משנה באשכולות

כשיוצרים אשכול של שירות מנוהל ל-Apache Kafka, מציינים לפחות רשת משנה אחת של Virtual Private Cloud ‏(VPC) שבה האשכול נגיש. במהלך יצירת האשכול, סוכן השירות של Managed Kafka מצרף את רשתות המשנה האלה לאשכול.

ההרשאה managedkafka.clusters.create מאפשרת לישות מורשית לצרף באופן עקיף כל תת-רשת שלסוכן השירות יש הרשאה להשתמש בה, גם אם לישות המורשית אין הרשאה ישירה להשתמש בתת-רשת.

מאפיינים של אשכול בשירות מנוהל ל-Apache Kafka

כשיוצרים או מעדכנים אשכול של שירות מנוהל ל-Apache Kafka, צריך לציין את המאפיינים הבאים.

שם האשכול

השם או המזהה של אשכול השירות המנוהל ל-Apache Kafka שאתם יוצרים. הנחיות למתן שמות לאשכולות מפורטות במאמר הנחיות למתן שמות למשאבים בשירות המנוהל ל-Apache Kafka. אי אפשר לשנות את השם של אשכול.

מיקום

המיקום שבו אתם יוצרים את האשכול. המיקום חייב להיות אחד מהאזורים הנתמכים של Google Cloud . אי אפשר לשנות את המיקום של האשכול אחר כך. רשימת המיקומים שבהם השירות המנוהל ל-Apache Kafka זמין

הגדרת הקיבולת

קיבולת המחשוב, הזיכרון והאחסון של אשכול Kafka. כשיוצרים אשכול, צריך לציין את הערכים הבאים:

• ‫vCPUs: מספר המעבדים הווירטואליים באשכול. צריך לפחות 3 מעבדי vCPU לכל אשכול.

• ‫Memory: נפח הזיכרון שמוקצה לאשכול. צריך להקצות בין ‎1GiB ל-‎8GiB לכל vCPU.

  לדוגמה, אם יוצרים אשכול עם 6 מעבדי vCPU, הזיכרון המינימלי שאפשר להקצות לאשכול הוא 6GiB (‏1GiB לכל מעבד vCPU), והמקסימלי הוא 48GiB (‏8GiB לכל מעבד vCPU).

השירות מקצה באופן אוטומטי ברוקרים, כולל אמצעי אחסון מסוג Persistent Disk לברוקרים. אפשר גם להפעיל את האפשרות Custom local storage, שמאפשרת לציין את קיבולת הדיסק של כל ברוקר. מידע נוסף זמין במאמר הגדרת גודל הדיסק של המתווך.

במאמר תכנון הגודל של אשכול Kafka מוסבר איך להעריך את הקיבולת הנדרשת לאשכול.

הגדרת רשת

הגדרת הרשת היא רשימה של רשתות משנה של VPC שמהן אפשר לגשת לאשכול. כדי ליצור או לצרוך הודעות, הלקוחות צריכים להיות מסוגלים להגיע לאחת מתת-הרשתות האלה.

ריכזנו כאן כמה הנחיות להגדרת הרשת:

• באשכול צריך להיות לפחות רשת משנה אחת. המספר המקסימלי הוא 10.

• מותר להשתמש ברשת משנה אחת בלבד לכל רשת באשכול נתון.

• כל תת-רשת צריכה להיות באותו אזור של האשכול. הפרויקט והרשת יכולים להיות שונים.

• כתובות ה-IP של ה-brokers ושל שרת האתחול מוקצות באופן אוטומטי בכל רשת משנה. בנוסף, ערכי DNS לכתובות ה-IP האלה נוצרים ברשתות ה-VPC המתאימות.

• אם מוסיפים רשת משנה מפרויקט אחר, צריך לתת הרשאות לחשבון השירות שמנוהל על ידי Google ומשויך לאשכול. למידע נוסף, קראו את המאמר חיבור אשכול בין פרויקטים.

אחרי שיוצרים את האשכול, אפשר לעדכן את רשימת רשתות המשנה. מידע נוסף על רשתות זמין במאמר הגדרת רשתות לשירות מנוהל ל-Apache Kafka.

סוכן השירות של Managed Kafka משתמש ברשתות המשנה של האשכול. לסוכן יש הרשאה להשתמש ברשתות המשנה באמצעות תפקיד ה-IAM סוכן שירות מנוהל של Kafka (roles/managedkafka.serviceAgent).

אשכולות ציבוריים

אתם יכולים להגדיר את האשכול כאשכול ציבורי כדי לאפשר לאפליקציות לקוח להתחבר לאשכול ולהריץ פעולות של Kafka (כולל שליחה וקבלה של הודעות) דרך האינטרנט הציבורי. כדי לאפשר גישה ציבורית, צריך לציין לפחות טווח אחד של כתובות IP מותרות.

כשמגדירים טווחי כתובות IP מותרים של מקורות, חלים הכללים הבאים:

• כל טווחי כתובות ה-IP המותרים של המקור חייבים להיות מצוינים בסימון IPv4 CIDR (לדוגמה, 203.0.113.0/24 או 198.51.100.5/32).

• כל גודל של רשת משנה ב-CIDR חייב להיות בין /16 ל-/32.

• טווחי ה-CIDR לא יכולים להיות חופפים.

• כל טווחי כתובות ה-IP המותרים של המקור חייבים להיות ניתנים לניתוב באופן ציבורי. טווחים של כתובות IP פרטיות, כמו 10.0.0.0/8 (RFC 1918), נדחים.

• אפשר לציין עד 500 טווחי כתובות IP מותרים של מקורות.

• אין תמיכה בכתובות IPv6.

אזהרה: כדי להגן על האשכול, צריך לאפשר גישה רק מטווחים מהימנים של כתובות IP חיצוניות. אל תחשפו את האשכול לטווחים לא מהימנים.

לפני שאתם מפעילים גישה ציבורית לאשכול, מומלץ לקרוא את המאמר חיבור לקוחות לאשכול ציבורי כדי להבין איך התכונה של גישה ציבורית מיושמת.

תוויות

תוויות הן צמדי מפתח/ערך שעוזרים לכם לארגן ולזהות את המשאבים. תוויות מאפשרות לסווג משאבים לפי סביבה. לדוגמה, "env:production" ו-"owner:data-engineering".

אתם יכולים לסנן ולחפש משאבים לפי התוויות שלהם. לדוגמה, נניח שיש לכם כמה אשכולות של שירות מנוהל ל-Apache Kafka במחלקות שונות. כדי למצוא את האשכול הרלוונטי, אפשר להגדיר ולחפש אשכולות עם התווית "department:marketing".

הגדרת האיזון מחדש

ההגדרה הזו קובעת אם השירות יאזן מחדש באופן אוטומטי את הרפליקות של המחיצות בין הברוקרים.

המצבים הזמינים הם:

• ‫Auto rebalance on scale up: כשהאפשרות הזו מופעלת, השירות מפעיל באופן אוטומטי איזון מחדש של הרפליקות כשמגדילים את האשכול. המצב הזה עוזר לשמור על חלוקת עומסים אחידה, אבל הוא עלול להשפיע באופן זמני על הביצועים במהלך פעולת האיזון מחדש.

• No rebalance: כשהאפשרות הזו מופעלת, השירות לא מאזן מחדש את הרפליקות באופן אוטומטי.

מידע נוסף זמין במאמר איזון מחדש של מחיצות ב-broker.

הצפנה

השירות המנוהל ל-Apache Kafka יכול להצפין הודעות באמצעותGoogle-owned and Google-managed encryption keys (ברירת המחדל) או מפתחות הצפנה בניהול הלקוח (CMEK). כל ההודעות מוצפנות גם כשהן נשמרות וגם כשהן נשלחות. אי אפשר לשנות את סוג ההצפנה של אשכול.

Google-owned and Google-managed encryption keys משמשים כברירת מחדל. המפתחות האלה נוצרים, מנוהלים ומאוחסנים על ידי Google Cloud בתשתית שלה.

מפתחות CMEK הם מפתחות הצפנה שאתם מנהלים באמצעות Cloud Key Management Service. התכונה הזו מאפשרת לכם לשלוט טוב יותר במפתחות שמשמשים להצפנת נתונים באחסון בשירותי Google Cloud נתמכים. השימוש ב-CMEK כרוך בעלויות נוספות שקשורות ל-Cloud Key Management Service. כדי להשתמש ב-CMEK, אוסף המפתחות צריך להיות באותו מיקום שבו נמצאים המשאבים שאתם משתמשים בהם. מידע נוסף זמין במאמר הגדרת הצפנת הודעות.

הגדרת mTLS

אפשר להגדיר mTLS כשיטת אימות חלופית שמשתמשת באישורי לקוח. ההגדרות כוללות את הפרטים הבאים:

• מאגרי CA: רשימה של 1 עד 10 מאגרים של Certificate Authority Service (CAS) שהאשכול סומך עליהם לצורך אימות לקוח.

• כללי מיפוי של ישויות SSL: מאפיין אופציונלי אך מומלץ של מתווך ssl.principal.mapping.rules, שנועד לפשט שמות ארוכים של ישויות אישורים לשימוש ברשימות ACL של Kafka.

מידע נוסף על mTLS זמין במאמר הגדרת אימות mTLS.

יצירת אשכול

לפני שיוצרים אשכול, כדאי לקרוא את המאמר בנושא מאפייני אשכול.

יצירת אשכול נמשכת בדרך כלל 20-30 דקות.

כדי ליצור אשכול:

המסוף

1. נכנסים לדף Clusters במסוף Google Cloud .

   כניסה לדף Clusters

2. לוחצים על יצירה.

   נפתח הדף Create Kafka cluster.

3. בשדה Cluster name, מזינים מחרוזת.

   מידע נוסף על מתן שמות לאשכולות זמין במאמר הנחיות למתן שמות למשאבים בשירות מנוהל ל-Apache Kafka.

4. בשדה Location, מזינים מיקום נתמך.

   מידע נוסף על מיקומים נתמכים זמין במאמר מיקומים נתמכים של שירות מנוהל ל-Apache Kafka.

5. בקטע Capacity configuration, מזינים ערכים בשדות Memory ו-vCPUs.

   למידע נוסף על קביעת הגודל של אשכול, קראו את המאמר תכנון הגודל של אשכול Kafka.

6. זה שינוי אופציונלי. כדי לציין את קיבולת הדיסק של ה-broker:

   1. מעבירים את המתג Enable custom local storage size למצב מופעל.

      חשוב: אחרי שלוחצים על Save, אי אפשר להשבית שוב את ההגדרה הזו באשכול הזה.

   2. בשדה Custom local storage, מזינים את גודל הדיסק לכל ברוקר, ב-GiB.

   אם משאירים את המתג הפעלת גודל מותאם אישית של אחסון מקומי במצב מושבת, המערכת מקצה את קיבולת ברירת המחדל של הדיסק. מידע נוסף זמין במאמר הגדרת גודל הדיסק של המתווך.

7. בקטע Network configuration, מזינים את הפרטים הבאים:

   1. ‫Project: הפרויקט שבו נמצאת רשת המשנה. רשת המשנה צריכה להיות באותו אזור שבו נמצא האשכול, אבל יכול להיות שהיא תהיה בפרויקט אחר.
   2. רשת: הרשת שאליה מחוברת תת-הרשת.
   3. ‫Subnetwork: השם של תת-הרשת.
   4. ‫Subnet URI path: השדה הזה מאוכלס באופן אוטומטי. אפשר גם להזין את הנתיב של רשת המשנה כאן. השם של תת-הרשת צריך להיות בפורמט: projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.
   5. לוחצים על סיום.

8. אופציונלי: כדי להוסיף עוד רשתות משנה, לוחצים על Add a connected subnet.

   אפשר להוסיף עוד רשתות משנה, עד 10 לכל היותר.

9. אופציונלי: מאפשרים גישה ציבורית לאשכול.

   1. בקטע Public cluster configuration, בקטע Cluster can be reached, בוחרים באפשרות In a private network and using public internet.

   2. בשדה רשימה מופרדת בפסיקים של טווחי IPv4 מקור מותרים בפורמט CIDR, מזינים את טווחי ה-IP של אפליקציות הלקוח בפורמט IPv4 CIDR (לדוגמה, 203.0.113.0/24). אין תמיכה בכתובות IPv6.

10. אופציונלי: מגדירים מפתח הצפנה בניהול הלקוח (CMEK).

    1. בקטע Encryption, בוחרים באפשרות Cloud KMS key.

    2. בשדה Key type, בוחרים באפשרות Cloud KMS.

    3. בשדה Select a customer-managed key, בוחרים או מזינים את ה-CMEK.

       אם האפשרות הזו לא מופיעה, ודאו שהפעלתם את Cloud KMS API בפרויקט.

    מידע נוסף זמין במאמר הגדרת הצפנת הודעות.

11. לוחצים על יצירה.

gcloud

1. מפעילים את Cloud Shell במסוף Google Cloud .

   הפעלת Cloud Shell

   בחלק התחתון של מסוף Google Cloud יתחיל סשן של Cloud Shell ותופיע הודעה של שורת הפקודה. Cloud Shell היא סביבת מעטפת שבה ה-CLI של Google Cloud מותקן ומוגדרים ערכים לפרויקט הקיים. הסשן יופעל תוך כמה שניות.

2. לפני השימוש בנתוני הפקודה הבאים, צריך להחליף את הנתונים הבאים:

   • ‫PROJECT_ID: מזהה הפרויקט.
   • ‫LOCATION: המיקום של האשכול.
   • ‫CLUSTER_ID: מזהה האשכול.
   • ‫CPU_COUNT: מספר ה-vCPU של האשכול.
   • ‫MEMORY: נפח הזיכרון של האשכול. דוגמה: 3GiB.
   • ‫SUBNET_ID: המזהה של תת-הרשת שאליה רוצים להתחבר. דוגמה: default.
   • ‫ENCRYPTION_KEY: המזהה של מפתח ההצפנה בניהול הלקוח שבו רוצים להשתמש באשכול. הפעולה הזאת אופציונלית.
   • ‫LABELS: התוויות שרוצים לשייך לאשכול.
   • ‫ALLOWED_SOURCE_IP_RANGES: טווחי כתובות ה-IP המותרים של המקור לאשכול הציבורי. הפעולה הזאת אופציונלית.

   מריצים את הפקודה הבאה:

   ‫Linux,‏ macOS או Cloud Shell

   הערה: חשוב לוודא שהפעלתם את Google Cloud CLI עם אימות ופרויקט, על ידי הרצת הפקודה gcloud init או gcloud auth login ו- gcloud config set project.

   gcloud managed-kafka clusters create CLUSTER_ID \
       --location=LOCATION \
       --cpu=CPU_COUNT \
       --memory=MEMORY \
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID \
       --auto-rebalance \
       --encryption-key=ENCRYPTION_KEY \
       --async \
       --labels=LABELS \
       --public-cluster \
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   ‏Windows (PowerShell)

   הערה: חשוב לוודא שהפעלתם את Google Cloud CLI עם אימות ופרויקט, על ידי הרצת הפקודה gcloud init או gcloud auth login ו- gcloud config set project.

   gcloud managed-kafka clusters create CLUSTER_ID `
       --location=LOCATION `
       --cpu=CPU_COUNT `
       --memory=MEMORY `
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID `
       --auto-rebalance `
       --encryption-key=ENCRYPTION_KEY `
       --async `
       --labels=LABELS `
       --public-cluster `
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Windows‏ (cmd.exe)

   הערה: חשוב לוודא שהפעלתם את Google Cloud CLI עם אימות ופרויקט, על ידי הרצת הפקודה gcloud init או gcloud auth login ו- gcloud config set project.

   gcloud managed-kafka clusters create CLUSTER_ID ^
       --location=LOCATION ^
       --cpu=CPU_COUNT ^
       --memory=MEMORY ^
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
       --auto-rebalance ^
       --encryption-key=ENCRYPTION_KEY ^
       --async ^
       --labels=LABELS ^
       --public-cluster ^
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   אמורים לקבל תגובה שדומה לזו:

   done: false
   metadata:
     '@type': type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata
     apiVersion: v1
     createTime: 'CREATE_TIME'
     requestedCancellation: false
     target: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID
     verb: create
   name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID
   

   שומרים את OPERATION_ID כדי לעקוב אחרי progress.

REST

לפני שמשתמשים בנתוני הבקשה, צריך להחליף את הנתונים הבאים:

• ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud
• ‫LOCATION: המיקום של האשכול.
• ‫CLUSTER_ID: מזהה האשכול.
• ‫CPU_COUNT: מספר ה-vCPU באשכול.
• ‫MEMORY: נפח הזיכרון של האשכול, בבייטים. דוגמה: 3221225472.
• ‫SUBNET_ID: מזהה תת-הרשת שאליה רוצים להתחבר. דוגמה: default.

ה-method של ה-HTTP וכתובת ה-URL:

POST https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID

תוכן בקשת JSON:

{
  "capacityConfig": {
    "vcpuCount": CPU_COUNT,
    "memoryBytes": MEMORY
  },
  "gcpConfig": {
    "accessConfig": {
      "networkConfigs": [
        {
          "subnet": "projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID"
        }
      ]
    }
  }
}

כדי לשלוח את הבקשה צריך להרחיב אחת מהאפשרויות הבאות:

‫Curl (Linux,‏ macOS או Cloud Shell)

הערה: הפקודה הבאה מבוססת על ההנחה שנכנסתם ל-CLI של gcloud באמצעות חשבון המשתמש שלכם, על ידי הרצת gcloud init או gcloud auth login, או באמצעות Cloud Shell שמחבר אתכם אוטומטית ל-CLI של gcloud. כדי לבדוק איזה חשבון פעיל, אפשר להריץ את הפקודה gcloud auth list.

שומרים את גוף הבקשה בקובץ בשם request.json ומריצים את הפקודה הבאה:

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID"

‎PowerShell (Windows)

הערה: הפקודה הבאה מבוססת על ההנחה שנכנסתם ל-CLI של gcloud באמצעות חשבון המשתמש שלכם, על ידי הרצת gcloud init או gcloud auth login. כדי לבדוק איזה חשבון פעיל, אפשר להריץ את הפקודה gcloud auth list.

שומרים את גוף הבקשה בקובץ בשם request.json ומריצים את הפקודה הבאה:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID" | Select-Object -Expand Content

אתם אמורים לקבל תגובת JSON שדומה לזו:

{
  "name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
  "metadata": {
    "@type": "type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata",
    "createTime": "CREATE_TIME",
    "target": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID",
    "verb": "create",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

Terraform

אתם יכולים להשתמש במשאב של Terraform כדי ליצור אשכול.

resource "google_managed_kafka_cluster" "default" {
  project    = data.google_project.default.project_id # Replace this with your project ID in quotes
  cluster_id = "my-cluster-id"
  location   = "us-central1"
  capacity_config {
    vcpu_count   = 3
    memory_bytes = 3221225472
  }
  gcp_config {
    access_config {
      network_configs {
        subnet = google_compute_subnetwork.default.id
      }
    }
  }
}

כדי ללמוד איך להחיל הגדרות ב-Terraform או להסיר אותן, ראו פקודות בסיסיות ב-Terraform.

המשך

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי ההוראות להגדרת Go בקטע התקנת ספריות הלקוח. מידע נוסף מופיע במאמרי העזרה של ה-API של Go עבור שירות מנוהל ל-Apache Kafka.

כדי לבצע אימות בשירות המנוהל ל-Apache Kafka, צריך להגדיר את Application Default Credentials ‏(ADC). מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func createCluster(w io.Writer, projectID, region, clusterID, subnet string, cpu, memoryBytes int64, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// subnet := "projects/my-project-id/regions/us-central1/subnetworks/default"
	// cpu := 3
	// memoryBytes := 3221225472
	ctx := context.Background()
	client, err := managedkafka.NewClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewClient got err: %w", err)
	}
	defer client.Close()

	locationPath := fmt.Sprintf("projects/%s/locations/%s", projectID, region)
	clusterPath := fmt.Sprintf("%s/clusters/%s", locationPath, clusterID)

	// Memory must be between 1 GiB and 8 GiB per CPU.
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   cpu,
		MemoryBytes: memoryBytes,
	}
	var networkConfig []*managedkafkapb.NetworkConfig
	networkConfig = append(networkConfig, &managedkafkapb.NetworkConfig{
		Subnet: subnet,
	})
	platformConfig := &managedkafkapb.Cluster_GcpConfig{
		GcpConfig: &managedkafkapb.GcpConfig{
			AccessConfig: &managedkafkapb.AccessConfig{
				NetworkConfigs: networkConfig,
			},
		},
	}
	rebalanceConfig := &managedkafkapb.RebalanceConfig{
		Mode: managedkafkapb.RebalanceConfig_AUTO_REBALANCE_ON_SCALE_UP,
	}
	cluster := &managedkafkapb.Cluster{
		Name:            clusterPath,
		CapacityConfig:  capacityConfig,
		PlatformConfig:  platformConfig,
		RebalanceConfig: rebalanceConfig,
	}

	req := &managedkafkapb.CreateClusterRequest{
		Parent:    locationPath,
		ClusterId: clusterID,
		Cluster:   cluster,
	}
	op, err := client.CreateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 10-40 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created cluster: %s\n", resp.Name)
	return nil
}

Java

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההתקנה של Java בקטע התקנת ספריות הלקוח. מידע נוסף מופיע במאמרי העזרה של Managed Service for Apache Kafka Java API.

כדי לבצע אימות בשירות המנוהל ל-Apache Kafka, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

import com.google.api.gax.longrunning.OperationFuture;
import com.google.api.gax.longrunning.OperationSnapshot;
import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.gax.retrying.RetryingFuture;
import com.google.api.gax.retrying.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.AccessConfig;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.CreateClusterRequest;
import com.google.cloud.managedkafka.v1.GcpConfig;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.NetworkConfig;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.RebalanceConfig;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateCluster {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String clusterId = "my-cluster";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    int cpu = 3;
    long memoryBytes = 3221225472L; // 3 GiB
    createCluster(projectId, region, clusterId, subnet, cpu, memoryBytes);
  }

  public static void createCluster(
      String projectId, String region, String clusterId, String subnet, int cpu, long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig =
        CapacityConfig.newBuilder().setVcpuCount(cpu).setMemoryBytes(memoryBytes).build();
    NetworkConfig networkConfig = NetworkConfig.newBuilder().setSubnet(subnet).build();
    GcpConfig gcpConfig =
        GcpConfig.newBuilder()
            .setAccessConfig(AccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
            .build();
    RebalanceConfig rebalanceConfig =
        RebalanceConfig.newBuilder()
            .setMode(RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP)
            .build();
    Cluster cluster =
        Cluster.newBuilder()
            .setCapacityConfig(capacityConfig)
            .setGcpConfig(gcpConfig)
            .setRebalanceConfig(rebalanceConfig)
            .build();

    // Create the settings to configure the timeout for polling operations
    ManagedKafkaSettings.Builder settingsBuilder = ManagedKafkaSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.createClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {

      CreateClusterRequest request =
          CreateClusterRequest.newBuilder()
              .setParent(LocationName.of(projectId, region).toString())
              .setClusterId(clusterId)
              .setCluster(cluster)
              .build();

      // The duration of this operation can vary considerably, typically taking between 10-40
      // minutes.
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.createClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf("Cluster creation started. Operation name: %s\nDone: %s\nMetadata: %s\n",
          operation.getName(),
          operation.isDone(),
          future.getMetadata().get().toString());

      while (!future.isDone()) {
        // The pollingFuture gives us the most recent status of the operation
        RetryingFuture<OperationSnapshot> pollingFuture = future.getPollingFuture();
        OperationSnapshot currentOp = pollingFuture.getAttemptResult().get();
        System.out.printf("Polling Operation:\nName: %s\n Done: %s\n",
            currentOp.getName(),
            currentOp.isDone());
      }

      // NOTE: future.get() blocks completion until the operation is complete (isDone =  True)
      Cluster response = future.get();
      System.out.printf("Created cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.createCluster got err: %s", e.getMessage());
    }
  }
}

Python

לפני שתנסו את הדוגמה הזו, פעלו לפי הוראות ההגדרה של Python בקטע התקנת ספריות הלקוח. מידע נוסף מופיע במאמרי העזרה של ה-API בשפת Python של שירות מנוהל ל-Apache Kafka.

כדי לבצע אימות בשירות המנוהל ל-Apache Kafka, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

from google.api_core.exceptions import GoogleAPICallError
from google.cloud import managedkafka_v1

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# subnet = "projects/my-project-id/regions/us-central1/subnetworks/default"
# cpu = 3
# memory_bytes = 3221225472

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
cluster.capacity_config.vcpu_count = cpu
cluster.capacity_config.memory_bytes = memory_bytes
cluster.gcp_config.access_config.network_configs = [
    managedkafka_v1.NetworkConfig(subnet=subnet)
]
cluster.rebalance_config.mode = (
    managedkafka_v1.RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP
)

request = managedkafka_v1.CreateClusterRequest(
    parent=client.common_location_path(project_id, region),
    cluster_id=cluster_id,
    cluster=cluster,
)

try:
    operation = client.create_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    # The duration of this operation can vary considerably, typically taking 10-40 minutes.
    # We can set a timeout of 3000s (50 minutes).
    response = operation.result(timeout=3000)
    print("Created cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e.message}")

מעקב אחרי פעולת היצירה של האשכול

אפשר להריץ את הפקודה הבאה רק אם השתמשתם ב-CLI של gcloud כדי ליצור את האשכול.

• יצירת אשכול נמשכת בדרך כלל 20-30 דקות. כדי לעקוב אחרי ההתקדמות של יצירת האשכול, הפקודה gcloud managed-kafka clusters create משתמשת בפעולה ממושכת (LRO), שאפשר לעקוב אחריה באמצעות הפקודה הבאה:

  gcloud managed-kafka operations describe OPERATION_ID \
      --location=LOCATION
  

  מחליפים את מה שכתוב בשדות הבאים:

  • ‫OPERATION_ID עם הערך של מזהה הפעולה מהקטע הקודם.
  • ‫LOCATION עם הערך של המיקום מהקטע הקודם.

פתרון בעיות

למידע על פתרון שגיאות שקשורות ליצירת אשכול, ראו שגיאות ביצירת אשכול Kafka.

מה השלב הבא?

• מעקב אחרי האשכול.

• מעדכנים את האשכול.

• הצגת רשימה של כל האשכולות.

• הצגת האשכול.

• מחיקת אשכול.

• יוצרים נושא באשכול.

• הגדרת הרשת לשירות המנוהל ל-Apache Kafka

‫Apache Kafka®‎ הוא סימן מסחרי רשום של The Apache Software Foundation או של השותפים העצמאיים שלה בארצות הברית או במדינות אחרות.