Revert "k/proto: use log-level warn for missmatch crc"

src/v/cluster/rm_stm.cc

@@ -996,6 +996,11 @@ ss::future<result<kafka_result>> rm_stm::do_idempotent_replicate(
   ss::lw_shared_ptr<available_promise<>> enqueued,
   ssx::semaphore_units& units,
   producer_previously_known known_producer) {
+    if (rand() % 10 > 5) {
+        units.return_all();
+        vlog(_ctx_log.warn, "timed out: {}", bid);
+        co_return cluster::errc::timeout;
+    }
     // Check if the producer bumped the epoch and reset accordingly.
     if (bid.pid.epoch > producer->id().epoch()) {
         producer->reset_with_new_epoch(bid.pid.epoch);

src/v/kafka/protocol/kafka_batch_adapter.cc

@@ -122,7 +122,7 @@ void kafka_batch_adapter::verify_crc(int32_t expected_crc, iobuf_parser in) {
     if (unlikely((uint32_t)expected_crc != crc.value())) {
         valid_crc = false;
         vlog(
-          klog.warn,
+          klog.error,
           "Cannot validate Kafka record batch. Missmatching CRC. Expected:{}, "
           "Got:{}",
           expected_crc,

tests/java/verifiers/src/main/java/idempotency/App.java

@@ -94,7 +94,7 @@ void produceLoop() {
         String kv = Long.toString(counter);
         ProducerRecord<String, String> record
             = new ProducerRecord<>(this.params.topic, kv, kv);
-        idempotentProducer.send(record).get();
+        idempotentProducer.send(record);
       } catch (Exception e) {
         ex = e;
         logger.error("Exception in produce loop: ", e);

tests/rptest/tests/crc_failure_test.py

@@ -0,0 +1,36 @@
+# Copyright 2022 Redpanda Data, Inc.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.md
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0
+
+import time
+from rptest.services.cluster import cluster
+from rptest.clients.types import TopicSpec
+from rptest.tests.end_to_end import EndToEndTest
+from rptest.services.redpanda import RESTART_LOG_ALLOW_LIST, RedpandaService
+from rptest.transactions.verifiers.idempotency_load_generator import PausableIdempotentProducer
+
+
+class ShutdownTest(EndToEndTest):
+    @cluster(num_nodes=4, log_allow_list=RESTART_LOG_ALLOW_LIST)
+    def crc_failure_repro_test(self):
+        self.topic = TopicSpec(partition_count=1, replication_factor=3)
+        rp_conf = {
+            'enable_leader_balancer': False,
+            'auto_create_topics_enabled': True
+        }
+        self.redpanda = RedpandaService(self.test_context,
+                                        3,
+                                        extra_rp_conf=rp_conf)
+        self.redpanda.start()
+        workload_svc = PausableIdempotentProducer(self.test_context,
+                                                  self.redpanda)
+        workload_svc.start()
+        workload_svc.start_producer(self.topic.name,
+                                    self.topic.partition_count)
+        time.sleep(3 * 60)
+        workload_svc.stop_producer()