[CLI-3355] Fix some confluent kafka topic consume errors

[sgagniere]

Release Notes

Breaking Changes

• PLACEHOLDER

New Features

• PLACEHOLDER

Bug Fixes

• Fix a bug preventing users from consuming audit logs with confluent kafka topic consume
• Fix bugs preventing users from producing or consuming from a Cloud cluster while logged out using confluent kafka topic [produce | consume] --bootstrap

Checklist

• Leave this box unchecked if features are not yet available in production

What

• The assumption that a SR cluster always exists is false in the audit log environment, so this PR changes the call to get SR ID and endpoint to only happen if the key or value type is avro, protobuf, or jsonschema (for consume)
• For produce, only make a call to get the endpoint and ID if the user has not already provided the endpoint while logged out, to avoid a panic
• The GetDataplaneToken call will fail for users consuming from a cloud cluster using --bootstrap while logged out, so this PR adds a check that the user is logged in before attempting to get the token

References

Test & Review

Tested with the following.

• while logged in: confluent kafka topic consume -b confluent-audit-log-events
• while logged out:
  • Produce:
    • confluent kafka topic produce string-topic --bootstrap $ENDPOINT --api-key $CLUSTER_KEY --api-secret $CLUSTER_SECRET
    • confluent kafka topic produce avro-topic --value-format avro --schema 100004 --bootstrap $ENDPOINT --api-key $CLUSTER_KEY --api-secret $CLUSTER_SECRET --schema-registry-endpoint $SR_ENDPOINT --schema-registry-api-key $SR_KEY --schema-registry-api-secret $SR_SECRET
    • confluent kafka topic produce json-topic --value-format jsonschema --schema 100005 --bootstrap $ENDPOINT --api-key $CLUSTER_KEY --api-secret $CLUSTER_SECRET --schema-registry-endpoint $SR_ENDPOINT --schema-registry-api-key $SR_KEY --schema-registry-api-secret $SR_SECRET
    • confluent kafka topic produce protobuf-topic --value-format protobuf --schema 100007 --bootstrap $ENDPOINT --api-key $CLUSTER_KEY --api-secret $CLUSTER_SECRET --schema-registry-endpoint $SR_ENDPOINT --schema-registry-api-key $SR_KEY --schema-registry-api-secret $SR_SECRET
  • Consume:
    • confluent kafka topic consume confluent-audit-log-events -b --bootstrap $AUDIT_CLUSTER_ENDPOINT --api-key $AUDIT_KEY --api-secret $AUDIT_SECRET
    • confluent kafka topic consume string-topic -b --bootstrap $ENDPOINT --api-key $CLUSTER_KEY --api-secret $CLUSTER_SECRET
    • confluent kafka topic consume avro-topic -b --value-format avro --bootstrap $ENDPOINT --api-key $CLUSTER_KEY --api-secret $CLUSTER_SECRET --schema-registry-endpoint $SR_ENDPOINT --schema-registry-api-key $SR_KEY --schema-registry-api-secret $SR_SECRET
    • confluent kafka topic consume json-topic -b --value-format jsonschema --bootstrap $ENDPOINT --api-key $CLUSTER_KEY --api-secret $CLUSTER_SECRET --schema-registry-endpoint $SR_ENDPOINT --schema-registry-api-key $SR_KEY --schema-registry-api-secret $SR_SECRET
    • confluent kafka topic consume protobuf-topic -b --value-format protobuf --bootstrap $ENDPOINT --api-key $CLUSTER_KEY --api-secret $CLUSTER_SECRET --schema-registry-endpoint $SR_ENDPOINT --schema-registry-api-key $SR_KEY --schema-registry-api-secret $SR_SECRET

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[channingdong]

Can we also handle the auth.GetDataplaneToken(c.Context) in consumeOnPrem() and produceOnPrem()?

[channingdong]

Discussed with Steven, we'll track the CP specific issue in CLI-3360.

[channingdong]

Approved.