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

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

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

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

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

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

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

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

ההרשאות הנדרשות

כדי ליצור אשכול, צריך את ההרשאות הבאות:

  • יצירת אשכול: managedkafka.clusters.create

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

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

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

שם האשכול

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

מיקום

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

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

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

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

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

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

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

הנחיות להערכת הקיבולת הנדרשת לאשכול מפורטות במאמר תכנון הגודל של אשכול Kafka.

הגדרת רשת

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

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

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

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

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

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

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

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

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

אפשר גם להגדיר את האשכול כאשכול ציבורי כדי לאפשר לאפליקציות לקוח להתחבר לאשכול ולהפעיל פעולות של 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.

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

תוויות

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

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

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

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

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

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

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

מידע נוסף זמין במאמר בנושא Rebalancing broker partitions.

הצפנה

בשירות המנוהל ל-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. בשדה שם האשכול, מזינים מחרוזת.

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

  4. בקטע מיקום, מזינים מיקום נתמך.

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

  5. בקטע Capacity configuration (הגדרת קיבולת), מזינים ערכים בשדות Memory (זיכרון) ו-vCPUs (מעבדים וירטואליים).

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

  6. זה שינוי אופציונלי. כדי לציין את קיבולת הדיסק של הברוקר, מבצעים את הפעולות הבאות:

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

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

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

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

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

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

    אפשר להוסיף רשתות משנה נוספות, עד ערך מקסימלי של 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. בקטע הצפנה, בוחרים באפשרות מפתח Cloud KMS.

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

    3. בקטע Select a customer-managed key (בחירת מפתח בניהול הלקוח), בוחרים או מזינים את מפתח ה-CMEK.

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

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

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

gcloud

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

    הפעלת 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

    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)

    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)

    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"
        }
      ]
    }
  }
}

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

אתם אמורים לקבל תגובת 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®‎ הוא סימן מסחרי רשום של The Apache Software Foundation או של השותפים העצמאיים שלה בארצות הברית או במדינות אחרות.