CI Failure fix attempt

Author: Madhavan

agent-c3/src/main/java/com/datastax/oss/cdc/agent/PulsarMutationSender.java

@@ -106,6 +106,12 @@ public org.apache.avro.Schema getNativeSchema(String cql3Type) {
      */
     @Override
     public boolean isSupported(final AbstractMutation<CFMetaData> mutation) {
+        // Check if metadata is null (table may have been dropped)
+        if (mutation.metadata == null) {
+            log.warn("Table metadata is null for mutation key={}, table may have been dropped, skipping mutation", mutation.key());
+            return false;
+        }
+        
         if (!pkSchemas.containsKey(mutation.key())) {
             for (ColumnDefinition cm : mutation.metadata.primaryKeyColumns()) {
                 if (!avroNativeTypes.containsKey(cm.type.asCQL3Type().toString())) {

agent-c4/src/main/java/com/datastax/oss/cdc/agent/PulsarMutationSender.java

@@ -110,6 +110,12 @@ public org.apache.avro.Schema getNativeSchema(String cql3Type) {
      */
     @Override
     public boolean isSupported(final AbstractMutation<TableMetadata> mutation) {
+        // Check if metadata is null (table may have been dropped)
+        if (mutation.metadata == null) {
+            log.warn("Table metadata is null for mutation key={}, table may have been dropped, skipping mutation", mutation.key());
+            return false;
+        }
+        
         if (!pkSchemas.containsKey(mutation.key())) {
             for (ColumnMetadata cm : mutation.metadata.primaryKeyColumns()) {
                 if (!avroSchemaTypes.containsKey(cm.type.asCQL3Type().toString())) {

agent-dse4/src/main/java/com/datastax/oss/cdc/agent/PulsarMutationSender.java

@@ -111,6 +111,12 @@ public SchemaAndWriter getPkSchema(String key) {
      */
     @Override
     public boolean isSupported(final AbstractMutation<TableMetadata> mutation) {
+        // Check if metadata is null (table may have been dropped)
+        if (mutation.metadata == null) {
+            log.warn("Table metadata is null for mutation key={}, table may have been dropped, skipping mutation", mutation.key());
+            return false;
+        }
+        
         if (!pkSchemas.containsKey(mutation.key())) {
             for (ColumnMetadata cm : mutation.metadata.primaryKeyColumns()) {
                 if (!avroSchemaTypes.containsKey(cm.type.asCQL3Type().toString())) {