github-actions[bot] commented on code in PR #69391:
URL: https://github.com/apache/doris/pull/69391#discussion_r4358294944


##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -790,7 +809,9 @@ public void modifyProperties(AlterRoutineLoadCommand 
command) throws UserExcepti
                 throw new DdlException("Only supports modification of PAUSED 
jobs");
             }
 
-            modifyPropertiesInternal(jobProperties, dataSourceProperties);
+            Map<String, String> convertedProperties = 
prepareModifyProperties(dataSourceProperties);
+            applyProperties(jobProperties, dataSourceProperties);
+            setConvertedCustomProperties(convertedProperties);

Review Comment:
   [P1] Preserve datetime defaults when reusing the conversion cache
   
   For a job created with a datetime `property.kafka_default_offsets`, the 
first conversion removes `kafka_origin_default_offsets` but leaves the numeric 
`kafka_default_offsets` in the cache. A job-only ALTER (for example, changing 
`max_batch_rows`) returns that existing cache from `prepareModifyProperties()`, 
so this unconditional setter extracts the remaining millis string into 
`kafkaDefaultOffSet`. `convertedDefaultOffsetToLong()` then falls back to 
`OFFSET_END`. If a partition is added while the job is paused and the job 
resumes, the scheduler updates partition progress before `prepare()` rebuilds 
the cache, records that partition's end offset, and skips its records since the 
configured datetime. Extract defaults only from a freshly resolved map, or 
preserve the derived default when reusing the cache.



##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -805,21 +826,37 @@ private void convertOffset(KafkaDataSourceProperties 
dataSourceProperties) throw
         if (partitionOffsets.isEmpty()) {
             return;
         }
+        Map<String, String> properties;
+        String brokerListSnapshot;
+        String topicSnapshot;
+        writeLock();
+        try {
+            // Recovery leaves the conversion cache empty. Validate the 
effective
+            // properties before an ALTER offset lookup can contact an older 
BE.
+            convertCustomProperties(false);

Review Comment:
   [P2] Merge repairing ALTER properties before validating offset lookups
   
   An older paused job with `ssl.ca.location=/etc/ca.pem` can be migrated to 
`FILE:ca.pem`, but adding `kafka_partitions=0` and `kafka_offsets=100` to that 
same valid ALTER makes this call reject the old raw-path map. 
`modifyProperties()` invokes `convertOffset()` before 
`prepareModifyProperties()` merges the replacement, so the repair never reaches 
validation. This also occurs for explicit positive offsets, where 
`KafkaUtil.getRealOffsets()` needs no BE request at all. Prepare and validate 
the merged effective properties first, then use that safe snapshot for any 
required offset lookup while preserving atomic publication.



##########
fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaPropertyValidationTest.java:
##########
@@ -0,0 +1,413 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.load.routineload;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.cloud.rpc.MetaServiceProxy;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.Pair;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.common.util.KafkaUtil;

Review Comment:
   [P1] Import the existing Kafka utility and ALTER command classes
   
   This new test imports `org.apache.doris.common.util.KafkaUtil` and 
`org.apache.doris.nereids.trees.plans.commands.load.AlterRoutineLoadCommand`, 
but neither class exists in those packages on this head. The actual classes are 
`org.apache.doris.datasource.kafka.KafkaUtil` and 
`org.apache.doris.nereids.trees.plans.commands.AlterRoutineLoadCommand`, as 
used by `KafkaRoutineLoadJob`. Both types are referenced by these tests, so FE 
test compilation fails before any of the new validation/recovery tests can run. 
Correct both imports.



##########
fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaPropertyValidationTest.java:
##########
@@ -0,0 +1,413 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.load.routineload;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.cloud.rpc.MetaServiceProxy;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.Pair;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.common.util.KafkaUtil;
+import org.apache.doris.common.util.SmallFileMgr;
+import org.apache.doris.common.util.SmallFileMgr.SmallFile;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.load.routineload.kafka.KafkaDataSourceProperties;
+import org.apache.doris.load.routineload.kafka.KafkaProgress;
+import org.apache.doris.load.routineload.kafka.KafkaRoutineLoadJob;
+import org.apache.doris.nereids.parser.NereidsParser;
+import 
org.apache.doris.nereids.trees.plans.commands.load.AlterRoutineLoadCommand;
+import 
org.apache.doris.nereids.trees.plans.commands.load.CreateRoutineLoadCommand;
+import org.apache.doris.persist.AlterRoutineLoadJobOperationLog;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.persist.RoutineLoadOperation;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
+import org.apache.doris.thrift.BackendServiceConstants;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+public class KafkaPropertyValidationTest {
+    private KafkaDataSourceProperties properties(String key, String value, 
boolean alter) {
+        Map<String, String> source = new HashMap<>();
+        if (!alter) {
+            source.put("kafka_broker_list", "127.0.0.1:9092");
+            source.put("kafka_topic", "test_topic");
+        }
+        source.put("property." + key, value);
+        KafkaDataSourceProperties properties = new 
KafkaDataSourceProperties(source);
+        properties.setAlter(alter);
+        properties.setTimezone("UTC");
+        return properties;
+    }
+
+    @Test
+    public void testCreateAndAlterRejectExecutableAndUnknownProperties() {
+        for (boolean alter : new boolean[] {false, true}) {
+            for (String key : List.of("sasl.kerberos.kinit.cmd", 
"sasl.kerberos.keytab",
+                    "sasl.kerberos.min.time.before.relogin", 
"plugin.library.paths", "ssl.engine.location",
+                    "ssl.engine.id", "ssl.providers", 
"unknown.future.property", "sasl.jaas.config",
+                    "security.protocol\0suffix")) {
+                AnalysisException error = 
Assertions.assertThrows(AnalysisException.class,
+                        () -> properties(key, "sensitive-value", 
alter).analyze());
+                Assertions.assertTrue(error.getMessage().contains("is not 
allowed in Routine Load"), key);
+                
Assertions.assertFalse(error.getMessage().contains("sensitive-value"));
+            }
+        }
+    }
+
+    @Test
+    public void testRejectKerberosRefreshReenable() {
+        for (boolean alter : new boolean[] {false, true}) {
+            for (String value : List.of("1", "60000", "00", "0\0suffix", 
"FILE:interval")) {
+                Assertions.assertThrows(AnalysisException.class,
+                        () -> 
properties("sasl.kerberos.min.time.before.relogin", value, alter).analyze());
+            }
+        }
+    }
+
+    @Test
+    public void testCertificateFilesRequireManagedReferences() throws 
Exception {
+        for (boolean alter : new boolean[] {false, true}) {
+            for (String key : BackendServiceConstants.KAFKA_FILE_PROPERTIES) {
+                for (String value : List.of("/etc/ssl/private/key.pem", 
"relative.pem", "", "FILE:")) {
+                    AnalysisException error = 
Assertions.assertThrows(AnalysisException.class,
+                            () -> properties(key, value, alter).analyze());
+                    Assertions.assertTrue(error.getMessage().contains("CREATE 
FILE"));
+                }
+                KafkaDataSourceProperties properties = properties(key, 
"FILE:client.pem", alter);
+                properties.analyze();
+                Assertions.assertEquals("FILE:client.pem", 
properties.getCustomKafkaProperties().get(key));
+            }
+        }
+    }
+
+    @Test
+    public void testNormalConsumerPropertiesAndDefaultOffsets() throws 
Exception {
+        Map<String, String> accepted = Map.of(
+                "group.id", "routine-load-group",
+                "client.id", "routine-load-client",
+                "security.protocol", "SASL_SSL",
+                "sasl.mechanisms", "SCRAM-SHA-256",
+                "sasl.username", "user",
+                "sasl.password", "secret",
+                "isolation.level", "read_committed",
+                "fetch.max.bytes", "1048576",
+                "kafka_default_offsets", "OFFSET_BEGINNING");
+        for (boolean alter : new boolean[] {false, true}) {
+            for (Map.Entry<String, String> property : accepted.entrySet()) {
+                properties(property.getKey(), property.getValue(), 
alter).analyze();
+            }
+        }
+    }
+
+    @Test
+    public void testAwsPropertiesWithAndWithoutPrefix() throws Exception {
+        for (String prefix : List.of("", "property.")) {
+            Map<String, String> source = new HashMap<>();
+            source.put("kafka_broker_list", "127.0.0.1:9098");
+            source.put("kafka_topic", "test_topic");
+            source.put("property.security.protocol", "SASL_SSL");
+            source.put("property.sasl.mechanism", "OAUTHBEARER");
+            source.put(prefix + "aws.region", "us-east-1");
+            source.put(prefix + "aws.role_arn", 
"arn:aws:iam::123456789012:role/test");
+            KafkaDataSourceProperties properties = new 
KafkaDataSourceProperties(source);
+            properties.setTimezone("UTC");
+            properties.analyze();
+            Assertions.assertEquals("us-east-1", 
properties.getCustomKafkaProperties().get("aws.region"));
+        }
+    }
+
+    @Test
+    public void testDisableKerberosCommandForOlderBackends() throws Exception {
+        properties("sasl.kerberos.min.time.before.relogin", "0", 
false).analyze();
+        for (Map<String, String> custom : List.of(Map.<String, String>of(),
+                Map.of("group.id", "test", "sasl.kerberos.principal", 
"user;unused"),
+                Map.of("kafka_default_offsets", "OFFSET_BEGINNING",
+                        "kafka_origin_default_offsets", "OFFSET_BEGINNING"))) {
+            KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 
1L, 1L,
+                    "127.0.0.1:9092", "test_topic", UserIdentity.ADMIN);
+            Deencapsulation.setField(job, "customProperties", custom);
+            Deencapsulation.invoke(job, "convertCustomProperties", true);
+            Assertions.assertEquals("0", job.getConvertedCustomProperties()
+                    .get("sasl.kerberos.min.time.before.relogin"));
+        }
+    }
+
+    @Test
+    public void testPersistedPropertiesRejectedBeforeMetadataRequests() {
+        KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 
1L,
+                "127.0.0.1:9092", "test_topic", UserIdentity.ADMIN);
+        Deencapsulation.setField(job, "customProperties", 
Map.of("plugin.library.paths", "unused.so"));
+        // Even a previously populated conversion cache must not bypass 
validation.
+        Deencapsulation.setField(job, "convertedCustomProperties", 
Map.of("group.id", "cached"));
+        DdlException error = Assertions.assertThrows(DdlException.class,
+                () -> Deencapsulation.invoke(job, "convertCustomProperties", 
false));
+        Assertions.assertTrue(error.getMessage().contains("is not allowed in 
Routine Load"));
+    }
+
+    // Build an already analyzed, persisted ALTER from an older FE without 
applying
+    // today's CREATE/ALTER admission rules to the historical input.
+    private AlterRoutineLoadJobOperationLog historicalAlter(Map<String, 
String> custom,
+            String topic, String brokers, List<Pair<Integer, Long>> offsets) {
+        Map<String, String> source = new HashMap<>();
+        custom.forEach((key, value) -> source.put("property." + key, value));
+        source.put("kafka_topic", topic);
+        source.put("kafka_broker_list", brokers);
+        KafkaDataSourceProperties properties = new 
KafkaDataSourceProperties(source);
+        properties.setAlter(true);
+        Deencapsulation.setField(properties, "dataSourceType", "KAFKA");
+        Deencapsulation.setField(properties, "customKafkaProperties", custom);
+        Deencapsulation.setField(properties, "topic", topic);
+        Deencapsulation.setField(properties, "brokerList", brokers);
+        properties.setKafkaPartitionOffsets(offsets);
+        AlterRoutineLoadJobOperationLog log = new 
AlterRoutineLoadJobOperationLog(1L,
+                Map.of("max_batch_rows", "2000000"), properties);
+        return GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(log), 
AlterRoutineLoadJobOperationLog.class);
+    }
+
+    @Test
+    public void testReplayHistoricalAlterRestoresCompleteState() throws 
Exception {
+        for (boolean cloud : new boolean[] {false, true}) {
+            try (MockedStatic<Config> config = 
Mockito.mockStatic(Config.class);
+                    MockedStatic<Env> env = Mockito.mockStatic(Env.class);
+                    MockedStatic<MetaServiceProxy> metaService = 
Mockito.mockStatic(MetaServiceProxy.class);
+                    MockedStatic<KafkaUtil> kafka = 
Mockito.mockStatic(KafkaUtil.class)) {
+                config.when(Config::isCloudMode).thenReturn(cloud);
+                KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, 
"test_job", 1L, 2L,
+                        "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN);
+                Deencapsulation.setField(job, "progress", new 
KafkaProgress(Map.of(0, 5L, 1, 6L)));
+                job.prepare();
+
+                AlterRoutineLoadJobOperationLog log = historicalAlter(
+                        Map.of("sasl.kerberos.min.time.before.relogin", 
"60000"),
+                        "topic_b", "127.0.0.2:9092", List.of(Pair.of(0, 
100L)));
+                job.replayModifyProperties(log);
+
+                Assertions.assertEquals("topic_b", job.getTopic());
+                Assertions.assertEquals("127.0.0.2:9092", job.getBrokerList());
+                Assertions.assertEquals(Map.of(0, 100L),
+                        ((KafkaProgress) 
job.getProgress()).getOffsetByPartition());
+                Assertions.assertEquals(2000000L, job.getMaxBatchRows());
+                Assertions.assertEquals("60000", job.getCustomProperties()
+                        
.get("property.sasl.kerberos.min.time.before.relogin"));
+                
Assertions.assertTrue(job.getConvertedCustomProperties().isEmpty());
+
+                DdlException error = 
Assertions.assertThrows(DdlException.class, job::prepare);
+                Assertions.assertTrue(error.getMessage().contains("Kerberos 
tickets must be managed externally"));
+                Assertions.assertThrows(DdlException.class,
+                        () -> Deencapsulation.invoke(job, 
"getAllKafkaPartitions"));
+                Assertions.assertThrows(DdlException.class,
+                        () -> Deencapsulation.invoke(job, "convertOffset", 
log.getDataSourceProperties()));
+
+                // A later committed repair must retain topic B and its 
offsets.
+                job.replayModifyProperties(historicalAlter(
+                        Map.of("sasl.kerberos.min.time.before.relogin", "0"), 
"", "", List.of()));
+                job.prepare();
+                Assertions.assertEquals("topic_b", job.getTopic());
+                Assertions.assertEquals("127.0.0.2:9092", job.getBrokerList());
+                Assertions.assertEquals(Map.of(0, 100L),
+                        ((KafkaProgress) 
job.getProgress()).getOffsetByPartition());
+                Assertions.assertEquals("0", job.getConvertedCustomProperties()
+                        .get("sasl.kerberos.min.time.before.relogin"));
+                env.verifyNoInteractions();
+                metaService.verifyNoInteractions();
+                kafka.verifyNoInteractions();
+            }
+        }
+    }
+
+    @Test
+    public void testReplayDefersManagedFileResolution() throws Exception {
+        Env env = Mockito.mock(Env.class);
+        SmallFileMgr smallFileMgr = Mockito.mock(SmallFileMgr.class);
+        try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+            envStatic.when(Env::getCurrentEnv).thenReturn(env);
+            Mockito.when(env.getSmallFileMgr()).thenReturn(smallFileMgr);
+            Mockito.when(smallFileMgr.getSmallFile(1L, "kafka", "missing.pem", 
true))
+                    .thenThrow(new DdlException("missing managed file"));
+            KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 
1L, 2L,
+                    "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN);
+
+            // Partition 3 need not exist in the pre-replay progress. These 
offsets
+            // were checked when the ALTER was committed, not during recovery.
+            
job.replayModifyProperties(historicalAlter(Map.of("ssl.ca.location", 
"FILE:missing.pem"),
+                    "topic_b", "127.0.0.2:9092", List.of(Pair.of(3, 100L))));
+            Assertions.assertEquals("topic_b", job.getTopic());
+            Assertions.assertEquals(Map.of(3, 100L),
+                    ((KafkaProgress) 
job.getProgress()).getOffsetByPartition());
+            Assertions.assertEquals(2000000L, job.getMaxBatchRows());
+            envStatic.verifyNoInteractions();
+            Mockito.verifyNoInteractions(smallFileMgr);
+
+            DdlException error = Assertions.assertThrows(DdlException.class, 
job::prepare);
+            Assertions.assertTrue(error.getMessage().contains("missing managed 
file"));
+            
Assertions.assertTrue(job.getConvertedCustomProperties().isEmpty());
+        }
+    }
+
+    @Test
+    public void testSchedulerPausesReplayedJobWithRejectedProperties() throws 
Exception {
+        Env env = Mockito.mock(Env.class);
+        EditLog editLog = Mockito.mock(EditLog.class);
+        RoutineLoadManager manager = Mockito.mock(RoutineLoadManager.class);
+        try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class);
+                MockedStatic<KafkaUtil> kafka = 
Mockito.mockStatic(KafkaUtil.class)) {
+            envStatic.when(Env::getCurrentEnv).thenReturn(env);
+            Mockito.when(env.getEditLog()).thenReturn(editLog);
+            KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 
1L, 2L,
+                    "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN);
+            
job.replayModifyProperties(historicalAlter(Map.of("plugin.library.paths", 
"unused.so"),
+                    "topic_b", "127.0.0.2:9092", List.of()));
+            
Mockito.when(manager.getRoutineLoadJobByState(Mockito.anySet())).thenReturn(List.of(job));
+
+            Deencapsulation.invoke(new RoutineLoadScheduler(manager), 
"process");
+
+            Assertions.assertEquals(RoutineLoadJob.JobState.PAUSED, 
job.getState());
+            
Assertions.assertTrue(job.getPauseReason().getMsg().contains("plugin.library.paths"));
+            Assertions.assertEquals("topic_b", job.getTopic());
+            kafka.verifyNoInteractions();
+            
Mockito.verify(editLog).logOpRoutineLoadJob(Mockito.any(RoutineLoadOperation.class));
+        }
+    }
+
+    @Test
+    public void testLiveAlterPublishesOnlyPreparedProperties() throws 
Exception {
+        Env env = Mockito.mock(Env.class);
+        SmallFileMgr smallFileMgr = Mockito.mock(SmallFileMgr.class);
+        try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+            envStatic.when(Env::getCurrentEnv).thenReturn(env);
+            Mockito.when(env.getSmallFileMgr()).thenReturn(smallFileMgr);
+            Mockito.when(smallFileMgr.getSmallFile(1L, "kafka", "missing.pem", 
true))
+                    .thenThrow(new DdlException("missing managed file"));
+            KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 
1L, 2L,
+                    "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN);
+            Deencapsulation.setField(job, "state", 
RoutineLoadJob.JobState.PAUSED);
+            job.prepare();
+            Map<String, String> converted = new 
HashMap<>(job.getConvertedCustomProperties());
+            AlterRoutineLoadJobOperationLog log = 
historicalAlter(Map.of("ssl.ca.location", "FILE:missing.pem"),
+                    "topic_b", "127.0.0.2:9092", List.of());
+            AlterRoutineLoadCommand command = 
Mockito.mock(AlterRoutineLoadCommand.class);
+            
Mockito.when(command.getAnalyzedJobProperties()).thenReturn(log.getJobProperties());
+            
Mockito.when(command.getDataSourceProperties()).thenReturn(log.getDataSourceProperties());
+
+            Assertions.assertThrows(DdlException.class, () -> 
job.modifyProperties(command));
+            Assertions.assertEquals("topic_a", job.getTopic());
+            Assertions.assertEquals("127.0.0.1:9092", job.getBrokerList());
+            Assertions.assertTrue(job.getCustomProperties().isEmpty());
+            Assertions.assertEquals(converted, 
job.getConvertedCustomProperties());
+            Mockito.verify(env, Mockito.never()).getEditLog();
+
+            KafkaDataSourceProperties safe = 
properties("sasl.kerberos.min.time.before.relogin", "0", true);
+            safe.analyze();
+            Mockito.when(command.getDataSourceProperties()).thenReturn(safe);
+            EditLog editLog = Mockito.mock(EditLog.class);
+            Mockito.when(env.getEditLog()).thenReturn(editLog);
+            job.modifyProperties(command);
+            
Assertions.assertEquals(Map.of("property.sasl.kerberos.min.time.before.relogin",
 "0"),

Review Comment:
   [P2] Include the analyzed default offset in the ALTER expectation
   
   After correcting the imports, this assertion still fails: `safe.analyze()` 
calls `analyzeKafkaDefaultOffsetProperty()`, which inserts 
`kafka_default_offsets=OFFSET_END` even for this ALTER. `applyProperties()` 
copies that entry into the raw custom map, and `getCustomProperties()` returns 
it as `property.kafka_default_offsets`; only the converted map removes the 
bookkeeping key. The actual map therefore has two entries, while this assertion 
expects one. Include the inserted default-offset property, or assert the 
intended entry without comparing against an incomplete map.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to