מחיקת נושא בשירות מנוהל של Google Cloud ל-Apache Kafka

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

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

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

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

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

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

כדי למחוק נושא, נדרשות ההרשאות הבאות:

• כדי למחוק נושא: managedkafka.topics.delete

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

מחיקת נושא

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

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

כדי למחוק נושא, פועלים לפי השלבים הבאים:

המסוף

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

   מעבר אל Clusters

2. ברשימת האשכולות, לוחצים על האשכול שאליו משתייך הנושא שרוצים למחוק.

   הדף Cluster details נפתח. בדף פרטי האשכול, הנושאים מפורטים בכרטיסייה Resources.

3. לוחצים על הנושא שרוצים למחוק.

   ייפתח הדף פרטי הנושא.

4. לוחצים על מחיקה ומאשרים את הפעולה.

gcloud

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

   הפעלת Cloud Shell

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

2. מריצים את הפקודה gcloud managed-kafka topics delete:

   gcloud managed-kafka topics delete TOPIC_ID \
       --cluster=CLUSTER_ID \
       --location=LOCATION_ID
   

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

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

   • ‫TOPIC_ID: המזהה של הנושא שרוצים למחוק.
   • ‫CLUSTER_ID: המזהה של האשכול שבו נמצא הנושא.
   • ‫LOCATION_ID: המיקום של האשכול.

REST

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

• ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud
• ‫LOCATION: המיקום של האשכול
• ‫CLUSTER_ID: מזהה האשכול
• ‫TOPIC_ID: מזהה הנושא

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

DELETE https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID

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

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

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

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

curl -X DELETE \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID"

‎PowerShell (Windows)

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

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

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

Invoke-WebRequest `
    -Method DELETE `
    -Headers $headers `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID" | Select-Object -Expand Content

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

{}

המשך

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

כדי לבצע אימות לשירות המנוהל ל-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 deleteTopic(w io.Writer, projectID, region, clusterID, topicID string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// topicID := "my-topic"
	ctx := context.Background()
	client, err := managedkafka.NewClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewClient got err: %w", err)
	}
	defer client.Close()

	clusterPath := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", projectID, region, clusterID)
	topicPath := fmt.Sprintf("%s/topics/%s", clusterPath, topicID)
	req := &managedkafkapb.DeleteTopicRequest{
		Name: topicPath,
	}
	if err := client.DeleteTopic(ctx, req); err != nil {
		return fmt.Errorf("client.DeleteTopic got err: %w", err)
	}
	fmt.Fprint(w, "Deleted topic\n")
	return nil
}

Java

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

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

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.TopicName;
import java.io.IOException;

public class DeleteTopic {

  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 topicId = "my-topic";
    deleteTopic(projectId, region, clusterId, topicId);
  }

  public static void deleteTopic(String projectId, String region, String clusterId, String topicId)
      throws Exception {
    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
      // This operation is being handled synchronously.
      managedKafkaClient.deleteTopic(TopicName.of(projectId, region, clusterId, topicId));
      System.out.println("Deleted topic");
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaClient.deleteTopic got err: %s", e.getMessage());
    }
  }
}

Python

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

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

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

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# topic_id = "my-topic"

client = managedkafka_v1.ManagedKafkaClient()

topic_path = client.topic_path(project_id, region, cluster_id, topic_id)
request = managedkafka_v1.DeleteTopicRequest(name=topic_path)

try:
    client.delete_topic(request=request)
    print("Deleted topic")
except NotFound as e:
    print(f"Failed to delete topic {topic_id} with error: {e.message}")

מה השלב הבא?

• מעקב אחרי נושא

• איך רואים קבוצת צרכנים

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

• מחיקה של קבוצת צרכנים