Downgrade to 4.1 fails due to schema changes

ERROR 20:56:39 Exiting due to error while processing commit log during initialization.
org.apache.cassandra.db.commitlog.CommitLogReadHandler$CommitLogReadException: Unexpected error deserializing mutation; saved to /var/folders/h1/s_3p1x3s3hl0hltbpck67m0h0000gn/T/mutation4184214444767150092dat.  This may be caused by replaying a mutation against a table with the same name but incompatible schema.  Exception follows: java.lang.RuntimeException: Unknown column compaction_properties during deserialization
	at org.apache.cassandra.db.commitlog.CommitLogReader.readMutation(CommitLogReader.java:464)
	at org.apache.cassandra.db.commitlog.CommitLogReader.readSection(CommitLogReader.java:397)
	at org.apache.cassandra.db.commitlog.CommitLogReader.readCommitLogSegment(CommitLogReader.java:244)
	at org.apache.cassandra.db.commitlog.CommitLogReader.readCommitLogSegment(CommitLogReader.java:147)
	at org.apache.cassandra.db.commitlog.CommitLogReplayer.replayFiles(CommitLogReplayer.java:191)
	at org.apache.cassandra.db.commitlog.CommitLog.recoverFiles(CommitLog.java:223)
	at org.apache.cassandra.db.commitlog.CommitLog.recoverSegmentsOnDisk(CommitLog.java:204)

/*
 * 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.cassandra.distributed.upgrade;

import java.io.IOException;
import java.io.File;
import java.util.concurrent.atomic.AtomicBoolean;

import org.junit.Test;

import org.apache.cassandra.distributed.api.IUpgradeableInstance;

public class DowngradeTest extends UpgradeTestBase
{
    @Test
    public void test() throws Throwable
    {
        AtomicBoolean first = new AtomicBoolean(true);
        new TestCase()
        .nodes(1)
        .withConfig(c -> {
            if (first.compareAndSet(true, false))
                c.set("storage_compatibility_mode", "CASSANDRA_4");
        })
        .downgradeTo(v41)
        .setup(cluster -> {})
// Uncomment if you want to test what happens after reading the commit log, which fails right now
//        .runBeforeNodeRestart((cluster, nodeId) -> {
//            IUpgradeableInstance inst = cluster.get(nodeId);
//            File f = new File((String) inst.config().get("commitlog_directory"));
//            deleteRecursive(f);
//        })
        .runAfterClusterUpgrade(cluster -> {})
        .run();
    }

    private void deleteRecursive(File f)
    {
        if (f.isDirectory())
        {
            File[] children = f.listFiles();
            if (children != null)
            {
                for (File c : children)
                    deleteRecursive(c);
            }
        }
        f.delete();
    }
}

diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java b/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java
index 5ee8780204..b4111e3b44 100644
--- a/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java
@@ -226,6 +226,12 @@ public class UpgradeTestBase extends DistributedTestBase
             return this;
         }

+        public TestCase downgradeTo(Semver to)
+        {
+            upgrade.add(new TestVersions(versions.getLatest(CURRENT), Collections.singletonList(versions.getLatest(to))));
+            return this;
+        }
+
         /**
          * performs all supported upgrade paths that exist in between from and to that include the current version.
          * This call is equivalent to calling {@code upgradesTo(from, CURRENT).upgradesFrom(CURRENT, to)}.